hyperactor/

actor.rs

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 * All rights reserved.
 *
 * This source code is licensed under the BSD-style license found in the
 * LICENSE file in the root directory of this source tree.
 */

#![allow(dead_code)] // Allow until this is used outside of tests.

//! This module contains all the core traits required to define and manage actors.

use std::any::TypeId;
use std::fmt;
use std::fmt::Debug;
use std::future::Future;
use std::future::IntoFuture;
use std::pin::Pin;
use std::sync::Arc;
use std::time::SystemTime;

use async_trait::async_trait;
use enum_as_inner::EnumAsInner;
use futures::FutureExt;
use futures::future::BoxFuture;
use serde::Deserialize;
use serde::Serialize;
use tokio::sync::watch;
use tokio::task::JoinHandle;
use typeuri::Named;

use crate as hyperactor; // for macros
use crate::ActorRef;
use crate::Data;
use crate::Message;
use crate::RemoteMessage;
use crate::checkpoint::CheckpointError;
use crate::checkpoint::Checkpointable;
use crate::clock::Clock;
use crate::clock::RealClock;
use crate::context;
use crate::mailbox::MailboxError;
use crate::mailbox::MailboxSenderError;
use crate::mailbox::MessageEnvelope;
use crate::mailbox::PortHandle;
use crate::mailbox::Undeliverable;
use crate::mailbox::UndeliverableMessageError;
use crate::mailbox::log::MessageLogError;
use crate::message::Castable;
use crate::message::IndexedErasedUnbound;
use crate::proc::Context;
use crate::proc::Instance;
use crate::proc::InstanceCell;
use crate::proc::Ports;
use crate::proc::Proc;
use crate::reference::ActorId;
use crate::reference::Index;
use crate::supervision::ActorSupervisionEvent;

pub mod remote;

/// An Actor is an independent, asynchronous thread of execution. Each
/// actor instance has a mailbox, whose messages are delivered through
/// the method [`Actor::handle`].
///
/// Actors communicate with each other by way of message passing.
/// Actors are assumed to be _deterministic_: that is, the state of an
/// actor is determined by the set (and order) of messages it receives.
#[async_trait]
pub trait Actor: Sized + Send + 'static {
    /// Initialize the actor, after the runtime has been fully initialized.
    /// Init thus provides a mechanism by which an actor can reliably and always
    /// receive some initial event that can be used to kick off further
    /// (potentially delayed) processing.
    async fn init(&mut self, _this: &Instance<Self>) -> Result<(), anyhow::Error> {
        // Default implementation: no init.
        Ok(())
    }

    /// Cleanup things used by this actor before shutting down. Notably this function
    /// is async and allows more complex cleanup. Simpler cleanup can be handled
    /// by the impl Drop for this Actor.
    /// If err is not None, it is the error that this actor is failing with. Any
    /// errors returned by this function will be logged and ignored.
    /// If err is None, any errors returned by this function will be propagated
    /// as an ActorError.
    /// This function is not called if there is a panic in the actor, as the
    /// actor may be in an indeterminate state. It is also not called if the
    /// process is killed, there is no atexit handler or signal handler.
    async fn cleanup(
        &mut self,
        _this: &Instance<Self>,
        _err: Option<&ActorError>,
    ) -> Result<(), anyhow::Error> {
        // Default implementation: no cleanup.
        Ok(())
    }

    /// Spawn a child actor, given a spawning capability (usually given by [`Instance`]).
    /// The spawned actor will be supervised by the parent (spawning) actor.
    fn spawn(self, cx: &impl context::Actor) -> anyhow::Result<ActorHandle<Self>> {
        cx.instance().spawn(self)
    }

    /// Spawns this actor in a detached state, handling its messages
    /// in a background task. The returned handle is used to control
    /// the actor's lifecycle and to interact with it.
    ///
    /// Actors spawned through `spawn_detached` are not attached to a supervision
    /// hierarchy, and not managed by a [`Proc`].
    fn spawn_detached(self) -> Result<ActorHandle<Self>, anyhow::Error> {
        Proc::local().spawn("anon", self)
    }

    /// This method is used by the runtime to spawn the actor server. It can be
    /// used by actors that require customized runtime setups
    /// (e.g., dedicated actor threads), or want to use a custom tokio runtime.
    #[hyperactor::instrument_infallible]
    fn spawn_server_task<F>(future: F) -> JoinHandle<F::Output>
    where
        F: Future + Send + 'static,
        F::Output: Send + 'static,
    {
        tokio::spawn(future)
    }

    /// Handle actor supervision event. Return `Ok(true)`` if the event is handled here.
    async fn handle_supervision_event(
        &mut self,
        _this: &Instance<Self>,
        _event: &ActorSupervisionEvent,
    ) -> Result<bool, anyhow::Error> {
        // By default, the supervision event is not handled, caller is expected to bubble it up.
        Ok(false)
    }

    /// Default undeliverable message handling behavior.
    async fn handle_undeliverable_message(
        &mut self,
        cx: &Instance<Self>,
        envelope: Undeliverable<MessageEnvelope>,
    ) -> Result<(), anyhow::Error> {
        handle_undeliverable_message(cx, envelope)
    }

    /// If overridden, we will use this name in place of the
    /// ActorId for talking about this actor in supervision error
    /// messages.
    fn display_name(&self) -> Option<String> {
        None
    }
}

/// Default implementation of [`Actor::handle_undeliverable_message`]. Defined
/// as a free function so that `Actor` implementations that override
/// [`Actor::handle_undeliverable_message`] can fallback to this default.
pub fn handle_undeliverable_message<A: Actor>(
    cx: &Instance<A>,
    Undeliverable(envelope): Undeliverable<MessageEnvelope>,
) -> Result<(), anyhow::Error> {
    assert_eq!(envelope.sender(), cx.self_id());

    anyhow::bail!(UndeliverableMessageError::DeliveryFailure { envelope });
}

/// An actor that does nothing. It is used to represent "client only" actors,
/// returned by [`Proc::instance`].
#[async_trait]
impl Actor for () {}

impl Referable for () {}

impl Binds<()> for () {
    fn bind(_ports: &Ports<Self>) {
        // Binds no ports.
    }
}

/// A Handler allows an actor to handle a specific message type.
#[async_trait]
pub trait Handler<M>: Actor {
    /// Handle the next M-typed message.
    async fn handle(&mut self, cx: &Context<Self>, message: M) -> Result<(), anyhow::Error>;
}

/// We provide this handler to indicate that actors can handle the [`Signal`] message.
/// Its actual handler is implemented by the runtime.
#[async_trait]
impl<A: Actor> Handler<Signal> for A {
    async fn handle(&mut self, _cx: &Context<Self>, _message: Signal) -> Result<(), anyhow::Error> {
        unimplemented!("signal handler should not be called directly")
    }
}

/// This handler provides a default behavior when a message sent by
/// the actor to another is returned due to delivery failure.
#[async_trait]
impl<A: Actor> Handler<Undeliverable<MessageEnvelope>> for A {
    async fn handle(
        &mut self,
        cx: &Context<Self>,
        message: Undeliverable<MessageEnvelope>,
    ) -> Result<(), anyhow::Error> {
        self.handle_undeliverable_message(cx, message).await
    }
}

/// This handler enables actors to unbind the [IndexedErasedUnbound]
/// message, and forward the result to corresponding handler.
#[async_trait]
impl<A, M> Handler<IndexedErasedUnbound<M>> for A
where
    A: Handler<M>,
    M: Castable,
{
    async fn handle(
        &mut self,
        cx: &Context<Self>,
        msg: IndexedErasedUnbound<M>,
    ) -> anyhow::Result<()> {
        let message = msg.downcast()?.bind()?;
        Handler::handle(self, cx, message).await
    }
}

/// An `Actor` that can be spawned remotely.
///
/// Bounds explained:
/// - `Actor`: only actors may be remotely spawned.
/// - `Referable`: marks the type as eligible for typed remote
///   references (`ActorRef<A>`); required because remote spawn
///   ultimately hands back an `ActorId` that higher-level APIs may
///   re-type as `ActorRef<A>`.
/// - `Binds<Self>`: lets the runtime wire this actor's message ports
///   when it is spawned (the blanket impl calls `handle.bind::<Self>()`).
///
/// `gspawn` is a type-erased entry point used by the remote
/// spawn/registry machinery. It takes serialized params and returns
/// the new actor's `ActorId`; application code shouldn't call it
/// directly.
#[async_trait]
pub trait RemoteSpawn: Actor + Referable + Binds<Self> {
    /// The type of parameters used to instantiate the actor remotely.
    type Params: RemoteMessage;

    /// Creates a new actor instance given its instantiation parameters.
    async fn new(params: Self::Params) -> anyhow::Result<Self>;

    /// A type-erased entry point to spawn this actor. This is
    /// primarily used by hyperactor's remote actor registration
    /// mechanism.
    // TODO: consider making this 'private' -- by moving it into a non-public trait as in [`cap`].
    fn gspawn(
        proc: &Proc,
        name: &str,
        serialized_params: Data,
    ) -> Pin<Box<dyn Future<Output = Result<ActorId, anyhow::Error>> + Send>> {
        let proc = proc.clone();
        let name = name.to_string();
        Box::pin(async move {
            let params = bincode::deserialize(&serialized_params)?;
            let actor = Self::new(params).await?;
            let handle = proc.spawn(&name, actor)?;
            // We return only the ActorId, not a typed ActorRef.
            // Callers that hold this ID can interact with the actor
            // only via the serialized/opaque messaging path, which
            // makes it safe to export across process boundaries.
            //
            // Note: the actor itself is still `A`-typed here; we
            // merely restrict the *capability* we hand out to an
            // untyped identifier.
            //
            // This will be replaced by a proper export/registry
            // mechanism.
            Ok(handle.bind::<Self>().actor_id)
        })
    }

    /// The type ID of this actor.
    fn get_type_id() -> TypeId {
        TypeId::of::<Self>()
    }
}

/// If an actor implements Default, we use this as the
/// `RemoteSpawn` implementation, too.
#[async_trait]
impl<A: Actor + Referable + Binds<Self> + Default> RemoteSpawn for A {
    type Params = ();

    async fn new(_params: Self::Params) -> anyhow::Result<Self> {
        Ok(Default::default())
    }
}

#[async_trait]
impl<T> Checkpointable for T
where
    T: RemoteMessage + Clone,
{
    type State = T;
    async fn save(&self) -> Result<Self::State, CheckpointError> {
        Ok(self.clone())
    }

    async fn load(state: Self::State) -> Result<Self, CheckpointError> {
        Ok(state)
    }
}

/// Errors that occur while serving actors. Each error is associated
/// with the ID of the actor being served.
#[derive(Debug)]
pub struct ActorError {
    /// The ActorId for the actor that generated this error.
    pub actor_id: Box<ActorId>,
    /// The kind of error that occurred.
    pub kind: Box<ActorErrorKind>,
}

/// The kinds of actor serving errors.
#[derive(thiserror::Error, Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ActorErrorKind {
    /// Generic error with a formatted message.
    #[error("{0}")]
    Generic(String),

    /// An error that occurred while trying to handle a supervision event.
    #[error("{0} while handling {1}")]
    ErrorDuringHandlingSupervision(String, Box<ActorSupervisionEvent>),
    /// The actor did not attempt to handle
    #[error("{0}")]
    UnhandledSupervisionEvent(Box<ActorSupervisionEvent>),
}

impl ActorErrorKind {
    /// Error while processing actor, i.e., returned by the actor's
    /// processing method.
    pub fn processing(err: anyhow::Error) -> Self {
        // Unbox err from the anyhow err. Check if it is an ActorErrorKind object.
        // If it is directly use it as the new ActorError's ActorErrorKind.
        // This lets us directly pass the ActorErrorKind::UnhandledSupervisionEvent
        // up the handling infrastructure.
        err.downcast::<ActorErrorKind>()
            .unwrap_or_else(|err| Self::Generic(err.to_string()))
    }

    /// Unwound stracktrace of a panic.
    pub fn panic(err: anyhow::Error) -> Self {
        Self::Generic(format!("panic: {}", err))
    }

    /// Error during actor initialization.
    pub fn init(err: anyhow::Error) -> Self {
        Self::Generic(format!("initialization error: {}", err))
    }

    /// Error during actor cleanup.
    pub fn cleanup(err: anyhow::Error) -> Self {
        Self::Generic(format!("cleanup error: {}", err))
    }

    /// An underlying mailbox error.
    pub fn mailbox(err: MailboxError) -> Self {
        Self::Generic(err.to_string())
    }

    /// An underlying mailbox sender error.
    pub fn mailbox_sender(err: MailboxSenderError) -> Self {
        Self::Generic(err.to_string())
    }

    /// An underlying checkpoint error.
    pub fn checkpoint(err: CheckpointError) -> Self {
        Self::Generic(format!("checkpoint error: {}", err))
    }

    /// An underlying message log error.
    pub fn message_log(err: MessageLogError) -> Self {
        Self::Generic(format!("message log error: {}", err))
    }

    /// The actor's state could not be determined.
    pub fn indeterminate_state() -> Self {
        Self::Generic("actor is in an indeterminate state".to_string())
    }
}

impl ActorError {
    /// Create a new actor server error with the provided id and kind.
    pub(crate) fn new(actor_id: &ActorId, kind: ActorErrorKind) -> Self {
        Self {
            actor_id: Box::new(actor_id.clone()),
            kind: Box::new(kind),
        }
    }
}

impl fmt::Display for ActorError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "serving {}: ", self.actor_id)?;
        fmt::Display::fmt(&self.kind, f)
    }
}

impl std::error::Error for ActorError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        self.kind.source()
    }
}

impl From<MailboxError> for ActorError {
    fn from(inner: MailboxError) -> Self {
        Self {
            actor_id: Box::new(inner.actor_id().clone()),
            kind: Box::new(ActorErrorKind::mailbox(inner)),
        }
    }
}

impl From<MailboxSenderError> for ActorError {
    fn from(inner: MailboxSenderError) -> Self {
        Self {
            actor_id: Box::new(inner.location().actor_id().clone()),
            kind: Box::new(ActorErrorKind::mailbox_sender(inner)),
        }
    }
}

/// A collection of signals to control the behavior of the actor.
/// Signals are internal runtime control plane messages and should not be
/// sent outside of the runtime.
///
/// These messages are not handled directly by actors; instead, the runtime
/// handles the various signals.
#[derive(Clone, Debug, Serialize, Deserialize, typeuri::Named)]
pub enum Signal {
    /// Stop the actor, after draining messages.
    DrainAndStop,

    /// Stop the actor immediately.
    Stop,

    /// The direct child with the given PID was stopped.
    ChildStopped(Index),
}
wirevalue::register_type!(Signal);

impl fmt::Display for Signal {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Signal::DrainAndStop => write!(f, "DrainAndStop"),
            Signal::Stop => write!(f, "Stop"),
            Signal::ChildStopped(index) => write!(f, "ChildStopped({})", index),
        }
    }
}

/// The runtime status of an actor.
#[derive(
    Debug,
    Serialize,
    Deserialize,
    PartialEq,
    Eq,
    Clone,
    typeuri::Named,
    EnumAsInner
)]
pub enum ActorStatus {
    /// The actor status is unknown.
    Unknown,
    /// The actor was created, but not yet started.
    Created,
    /// The actor is initializing. It is not yet ready to receive messages.
    Initializing,
    /// The actor is in "client" state: the user is managing the actor's
    /// mailboxes manually.
    Client,
    /// The actor is ready to receive messages, but is currently idle.
    Idle,
    /// The actor has been processing a message, beginning at the specified
    /// instant. The message handler and arm is included.
    /// TODO: we shoudl use interned representations here, so we don't copy
    /// strings willy-nilly.
    Processing(SystemTime, Option<(String, Option<String>)>),
    /// The actor has been saving its state.
    Saving(SystemTime),
    /// The actor has been loading its state.
    Loading(SystemTime),
    /// The actor is stopping. It is draining messages.
    Stopping,
    /// The actor is stopped. It is no longer processing messages.
    Stopped,
    /// The actor failed with the provided actor error.
    Failed(ActorErrorKind),
}

impl ActorStatus {
    /// Tells whether the status is a terminal state.
    pub fn is_terminal(&self) -> bool {
        self.is_stopped() || self.is_failed()
    }

    /// Create a generic failure status with the provided error message.
    pub fn generic_failure(message: impl Into<String>) -> Self {
        Self::Failed(ActorErrorKind::Generic(message.into()))
    }

    fn span_string(&self) -> &'static str {
        self.arm().unwrap_or_default()
    }
}

impl fmt::Display for ActorStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Unknown => write!(f, "unknown"),
            Self::Created => write!(f, "created"),
            Self::Initializing => write!(f, "initializing"),
            Self::Client => write!(f, "client"),
            Self::Idle => write!(f, "idle"),
            Self::Processing(instant, None) => {
                write!(
                    f,
                    "processing for {}ms",
                    RealClock
                        .system_time_now()
                        .duration_since(*instant)
                        .unwrap_or_default()
                        .as_millis()
                )
            }
            Self::Processing(instant, Some((handler, None))) => {
                write!(
                    f,
                    "{}: processing for {}ms",
                    handler,
                    RealClock
                        .system_time_now()
                        .duration_since(*instant)
                        .unwrap_or_default()
                        .as_millis()
                )
            }
            Self::Processing(instant, Some((handler, Some(arm)))) => {
                write!(
                    f,
                    "{},{}: processing for {}ms",
                    handler,
                    arm,
                    RealClock
                        .system_time_now()
                        .duration_since(*instant)
                        .unwrap_or_default()
                        .as_millis()
                )
            }
            Self::Saving(instant) => {
                write!(
                    f,
                    "saving for {}ms",
                    RealClock
                        .system_time_now()
                        .duration_since(*instant)
                        .unwrap_or_default()
                        .as_millis()
                )
            }
            Self::Loading(instant) => {
                write!(
                    f,
                    "loading for {}ms",
                    RealClock
                        .system_time_now()
                        .duration_since(*instant)
                        .unwrap_or_default()
                        .as_millis()
                )
            }
            Self::Stopping => write!(f, "stopping"),
            Self::Stopped => write!(f, "stopped"),
            Self::Failed(err) => write!(f, "failed: {}", err),
        }
    }
}

/// ActorHandles represent a (local) serving actor. It is used to access
/// its messaging and signal ports, as well as to synchronize with its
/// lifecycle (e.g., providing joins).  Once dropped, the handle is
/// detached from the underlying actor instance, and there is no longer
/// any way to join it.
///
/// Correspondingly, [`crate::ActorRef`]s refer to (possibly) remote
/// actors.
pub struct ActorHandle<A: Actor> {
    cell: InstanceCell,
    ports: Arc<Ports<A>>,
}

/// A handle to a running (local) actor.
impl<A: Actor> ActorHandle<A> {
    pub(crate) fn new(cell: InstanceCell, ports: Arc<Ports<A>>) -> Self {
        Self { cell, ports }
    }

    /// The actor's cell. Used primarily for testing.
    /// TODO: this should not be a public API.
    pub(crate) fn cell(&self) -> &InstanceCell {
        &self.cell
    }

    /// The [`ActorId`] of the actor represented by this handle.
    pub fn actor_id(&self) -> &ActorId {
        self.cell.actor_id()
    }

    /// Signal the actor to drain its current messages and then stop.
    pub fn drain_and_stop(&self) -> Result<(), ActorError> {
        tracing::info!("ActorHandle::drain_and_stop called: {}", self.actor_id());
        self.cell.signal(Signal::DrainAndStop)
    }

    /// A watch that observes the lifecycle state of the actor.
    pub fn status(&self) -> watch::Receiver<ActorStatus> {
        self.cell.status().clone()
    }

    /// Send a message to the actor. Messages sent through the handle
    /// are always queued in process, and do not require serialization.
    pub fn send<M: Message>(&self, message: M) -> Result<(), MailboxSenderError>
    where
        A: Handler<M>,
    {
        self.ports.get().send(message)
    }

    /// Return a port for the provided message type handled by the actor.
    pub fn port<M: Message>(&self) -> PortHandle<M>
    where
        A: Handler<M>,
    {
        self.ports.get()
    }

    /// TEMPORARY: bind...
    /// TODO: we shoudl also have a default binding(?)
    pub fn bind<R: Binds<A>>(&self) -> ActorRef<R> {
        self.cell.bind(self.ports.as_ref())
    }
}

/// IntoFuture allows users to await the handle to join it. The future
/// resolves when the actor itself has stopped processing messages.
/// The future resolves to the actor's final status.
impl<A: Actor> IntoFuture for ActorHandle<A> {
    type Output = ActorStatus;
    type IntoFuture = BoxFuture<'static, Self::Output>;

    fn into_future(self) -> Self::IntoFuture {
        let future = async move {
            let mut status_receiver = self.cell.status().clone();
            let result = status_receiver.wait_for(ActorStatus::is_terminal).await;
            match result {
                Err(_) => ActorStatus::Unknown,
                Ok(status) => status.clone(),
            }
        };

        future.boxed()
    }
}

impl<A: Actor> Debug for ActorHandle<A> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
        f.debug_struct("ActorHandle").field("cell", &"..").finish()
    }
}

impl<A: Actor> Clone for ActorHandle<A> {
    fn clone(&self) -> Self {
        Self {
            cell: self.cell.clone(),
            ports: self.ports.clone(),
        }
    }
}

/// `Referable` is a marker trait for types that can appear as
/// remote references across process boundaries.
///
/// It is not limited to concrete [`Actor`] implementations. For
/// example, façade types generated by [`behavior!`] implement
/// `Referable` so that you can hand out restricted or stable APIs
/// while still using the same remote messaging machinery.
///
/// Implementing this trait means the type can be identified (`Named`)
/// so the runtime knows what it is.
///
///  In contrast, [`RemoteSpawn`] is the trait that marks *actors*
/// that can actually be **spawned remotely**. A behavior may be a
/// `Referable` but is never a `RemoteSpawn`.
pub trait Referable: Named {}

/// Binds determines how an actor's ports are bound to a specific
/// reference type.
pub trait Binds<A: Actor>: Referable {
    /// Bind ports in this actor.
    fn bind(ports: &Ports<A>);
}

/// Handles is a marker trait specifying that message type [`M`]
/// is handled by a specific actor type.
pub trait RemoteHandles<M: RemoteMessage>: Referable {}

/// Check if the actor behaves-as the a given behavior (defined by [`behavior!`]).
///
/// ```
/// # use serde::Serialize;
/// # use serde::Deserialize;
/// # use typeuri::Named;
/// # use hyperactor::Actor;
///
/// // First, define a behavior, based on handling a single message type `()`.
/// hyperactor::behavior!(UnitBehavior, ());
///
/// #[derive(Debug, Default)]
/// struct MyActor;
///
/// impl Actor for MyActor {}
///
/// #[async_trait::async_trait]
/// impl hyperactor::Handler<()> for MyActor {
///     async fn handle(
///         &mut self,
///         _cx: &hyperactor::Context<Self>,
///         _message: (),
///     ) -> Result<(), anyhow::Error> {
///         // no-op
///         Ok(())
///     }
/// }
///
/// hyperactor::assert_behaves!(MyActor as UnitBehavior);
/// ```
#[macro_export]
macro_rules! assert_behaves {
    ($ty:ty as $behavior:ty) => {
        const _: fn() = || {
            fn check<B: hyperactor::actor::Binds<$ty>>() {}
            check::<$behavior>();
        };
    };
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;
    use std::time::Duration;

    use tokio::time::timeout;

    use super::*;
    use crate as hyperactor;
    use crate::Actor;
    use crate::OncePortHandle;
    use crate::PortRef;
    use crate::checkpoint::CheckpointError;
    use crate::checkpoint::Checkpointable;
    use crate::test_utils::pingpong::PingPongActor;
    use crate::test_utils::pingpong::PingPongMessage;
    use crate::test_utils::proc_supervison::ProcSupervisionCoordinator; // for macros

    #[derive(Debug)]
    struct EchoActor(PortRef<u64>);

    #[async_trait]
    impl Actor for EchoActor {}

    #[async_trait]
    impl Handler<u64> for EchoActor {
        async fn handle(&mut self, cx: &Context<Self>, message: u64) -> Result<(), anyhow::Error> {
            let Self(port) = self;
            port.send(cx, message)?;
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_server_basic() {
        let proc = Proc::local();
        let client = proc.attach("client").unwrap();
        let (tx, mut rx) = client.open_port();
        let actor = EchoActor(tx.bind());
        let handle = proc.spawn::<EchoActor>("echo", actor).unwrap();
        handle.send(123u64).unwrap();
        handle.drain_and_stop().unwrap();
        handle.await;

        assert_eq!(rx.drain(), vec![123u64]);
    }

    #[tokio::test]
    async fn test_ping_pong() {
        let proc = Proc::local();
        let client = proc.attach("client").unwrap();
        let (undeliverable_msg_tx, _) = client.open_port();

        let ping_actor = PingPongActor::new(Some(undeliverable_msg_tx.bind()), None, None);
        let pong_actor = PingPongActor::new(Some(undeliverable_msg_tx.bind()), None, None);
        let ping_handle = proc.spawn::<PingPongActor>("ping", ping_actor).unwrap();
        let pong_handle = proc.spawn::<PingPongActor>("pong", pong_actor).unwrap();

        let (local_port, local_receiver) = client.open_once_port();

        ping_handle
            .send(PingPongMessage(10, pong_handle.bind(), local_port.bind()))
            .unwrap();

        assert!(local_receiver.recv().await.unwrap());
    }

    #[tokio::test]
    async fn test_ping_pong_on_handler_error() {
        let proc = Proc::local();
        let client = proc.attach("client").unwrap();
        let (undeliverable_msg_tx, _) = client.open_port();

        // Need to set a supervison coordinator for this Proc because there will
        // be actor failure(s) in this test which trigger supervision.
        ProcSupervisionCoordinator::set(&proc).await.unwrap();

        let error_ttl = 66;

        let ping_actor =
            PingPongActor::new(Some(undeliverable_msg_tx.bind()), Some(error_ttl), None);
        let pong_actor =
            PingPongActor::new(Some(undeliverable_msg_tx.bind()), Some(error_ttl), None);
        let ping_handle = proc.spawn::<PingPongActor>("ping", ping_actor).unwrap();
        let pong_handle = proc.spawn::<PingPongActor>("pong", pong_actor).unwrap();

        let (local_port, local_receiver) = client.open_once_port();

        ping_handle
            .send(PingPongMessage(
                error_ttl + 1, // will encounter an error at TTL=66
                pong_handle.bind(),
                local_port.bind(),
            ))
            .unwrap();

        // TODO: Fix this receiver hanging issue in T200423722.
        #[allow(clippy::disallowed_methods)]
        let res: Result<Result<bool, MailboxError>, tokio::time::error::Elapsed> =
            timeout(Duration::from_secs(5), local_receiver.recv()).await;
        assert!(res.is_err());
    }

    #[derive(Debug)]
    struct InitActor(bool);

    #[async_trait]
    impl Actor for InitActor {
        async fn init(&mut self, _this: &Instance<Self>) -> Result<(), anyhow::Error> {
            self.0 = true;
            Ok(())
        }
    }

    #[async_trait]
    impl Handler<OncePortHandle<bool>> for InitActor {
        async fn handle(
            &mut self,
            _cx: &Context<Self>,
            port: OncePortHandle<bool>,
        ) -> Result<(), anyhow::Error> {
            port.send(self.0)?;
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_init() {
        let proc = Proc::local();
        let actor = InitActor(false);
        let handle = proc.spawn::<InitActor>("init", actor).unwrap();
        let client = proc.attach("client").unwrap();

        let (port, receiver) = client.open_once_port();
        handle.send(port).unwrap();
        assert!(receiver.recv().await.unwrap());

        handle.drain_and_stop().unwrap();
        handle.await;
    }

    #[derive(Debug)]
    struct CheckpointActor {
        // The actor does nothing but sum the values of messages.
        sum: u64,
        port: PortRef<u64>,
    }

    #[async_trait]
    impl Actor for CheckpointActor {}

    #[async_trait]
    impl Handler<u64> for CheckpointActor {
        async fn handle(&mut self, cx: &Context<Self>, value: u64) -> Result<(), anyhow::Error> {
            self.sum += value;
            self.port.send(cx, self.sum)?;
            Ok(())
        }
    }

    #[async_trait]
    impl Checkpointable for CheckpointActor {
        type State = (u64, PortRef<u64>);

        async fn save(&self) -> Result<Self::State, CheckpointError> {
            Ok((self.sum, self.port.clone()))
        }

        async fn load(state: Self::State) -> Result<Self, CheckpointError> {
            let (sum, port) = state;
            Ok(CheckpointActor { sum, port })
        }
    }

    type MultiValues = Arc<Mutex<(u64, String)>>;

    struct MultiValuesTest {
        proc: Proc,
        values: MultiValues,
        handle: ActorHandle<MultiActor>,
        client: Instance<()>,
        _client_handle: ActorHandle<()>,
    }

    impl MultiValuesTest {
        async fn new() -> Self {
            let proc = Proc::local();
            let values: MultiValues = Arc::new(Mutex::new((0, "".to_string())));
            let actor = MultiActor(values.clone());
            let handle = proc.spawn::<MultiActor>("myactor", actor).unwrap();
            let (client, client_handle) = proc.instance("client").unwrap();
            Self {
                proc,
                values,
                handle,
                client,
                _client_handle: client_handle,
            }
        }

        fn send<M>(&self, message: M)
        where
            M: RemoteMessage,
            MultiActor: Handler<M>,
        {
            self.handle.send(message).unwrap()
        }

        async fn sync(&self) {
            let (port, done) = self.client.open_once_port::<bool>();
            self.handle.send(port).unwrap();
            assert!(done.recv().await.unwrap());
        }

        fn get_values(&self) -> (u64, String) {
            self.values.lock().unwrap().clone()
        }
    }

    #[derive(Debug)]
    #[hyperactor::export(handlers = [u64, String])]
    struct MultiActor(MultiValues);

    #[async_trait]
    impl Actor for MultiActor {}

    #[async_trait]
    impl Handler<u64> for MultiActor {
        async fn handle(&mut self, _cx: &Context<Self>, message: u64) -> Result<(), anyhow::Error> {
            let mut vals = self.0.lock().unwrap();
            vals.0 = message;
            Ok(())
        }
    }

    #[async_trait]
    impl Handler<String> for MultiActor {
        async fn handle(
            &mut self,
            _cx: &Context<Self>,
            message: String,
        ) -> Result<(), anyhow::Error> {
            let mut vals = self.0.lock().unwrap();
            vals.1 = message;
            Ok(())
        }
    }

    #[async_trait]
    impl Handler<OncePortHandle<bool>> for MultiActor {
        async fn handle(
            &mut self,
            _cx: &Context<Self>,
            message: OncePortHandle<bool>,
        ) -> Result<(), anyhow::Error> {
            message.send(true).unwrap();
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_multi_handler_refs() {
        let test = MultiValuesTest::new().await;

        test.send(123u64);
        test.send("foo".to_string());
        test.sync().await;
        assert_eq!(test.get_values(), (123u64, "foo".to_string()));

        let myref: ActorRef<MultiActor> = test.handle.bind();

        myref.port().send(&test.client, 321u64).unwrap();
        test.sync().await;
        assert_eq!(test.get_values(), (321u64, "foo".to_string()));

        myref.port().send(&test.client, "bar".to_string()).unwrap();
        test.sync().await;
        assert_eq!(test.get_values(), (321u64, "bar".to_string()));
    }

    #[tokio::test]
    async fn test_ref_behavior() {
        let test = MultiValuesTest::new().await;

        test.send(123u64);
        test.send("foo".to_string());

        hyperactor::behavior!(MyActorBehavior, u64, String);

        let myref: ActorRef<MyActorBehavior> = test.handle.bind();
        myref.port().send(&test.client, "biz".to_string()).unwrap();
        myref.port().send(&test.client, 999u64).unwrap();

        test.sync().await;
        assert_eq!(test.get_values(), (999u64, "biz".to_string()));
    }

    #[tokio::test]
    async fn test_actor_handle_downcast() {
        #[derive(Debug, Default)]
        struct NothingActor;

        impl Actor for NothingActor {}

        // Just test that we can round-trip the handle through a downcast.

        let proc = Proc::local();
        let handle = proc.spawn("nothing", NothingActor).unwrap();
        let cell = handle.cell();

        // Invalid actor doesn't succeed.
        assert!(cell.downcast_handle::<EchoActor>().is_none());

        let handle = cell.downcast_handle::<NothingActor>().unwrap();
        handle.drain_and_stop().unwrap();
        handle.await;
    }
}