ActorStatus::Failed(String) -> Failed(ActorErrorKind) (#1815)

Summary:
Pull Request resolved: #1815

Refactor to get a more structured error to ActorStatus. This deletes the unused structure in ActorErrorKind to make it cloneable/serializable so that we can put it in ActorStatus.

Now that we can represent an ActorStatus that is an SupervisionEvent of a child, we can start to chain/collapse the supervision chain as it bubbles up.
ghstack-source-id: 322758617
exported-using-ghexport

Reviewed By: mariusae

Differential Revision: D86723978

fbshipit-source-id: 3587161f752536ff913cb20ef616bd7c00ae9396

Author: zdevito

controller/src/lib.rs

@@ -544,7 +544,7 @@ impl ControllerMessageHandler for ControllerActor {
                             (
                                 actor.clone(),
                                 match status {
-                                    ActorStatus::Failed(msg) => msg.clone(),
+                                    ActorStatus::Failed(msg) => msg.to_string(),
                                     _ => format!("unexpected actor status {status}"),
                                 },
                             )

docs/source/books/hyperactor-book/src/actors/actor_handle.md

@@ -162,7 +162,7 @@ impl<A: Actor> IntoFuture for ActorHandle<A> {
             let result = status_receiver.wait_for(ActorStatus::is_terminal).await;
             match result {
                 Err(_) => ActorStatus::Unknown,
-                Ok(status) => status.passthrough(),
+                Ok(status) => status.clone(),
             }
         };
         future.boxed()

hyperactor/src/actor.rs

@@ -315,49 +315,58 @@ pub struct ActorError {
 }
 
 /// The kinds of actor serving errors.
-#[derive(thiserror::Error, Debug)]
+#[derive(thiserror::Error, Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum ActorErrorKind {
+    /// Generic error with a formatted message.
+    #[error("{0}")]
+    Generic(String),
+
+    /// An actor supervision event was not handled.
+    #[error("supervision: {0}")]
+    UnhandledSupervisionEvent(Box<ActorSupervisionEvent>),
+}
+
+impl ActorErrorKind {
     /// Error while processing actor, i.e., returned by the actor's
     /// processing method.
-    #[error("processing error: {0}")]
-    Processing(#[source] anyhow::Error),
+    pub fn processing(err: anyhow::Error) -> Self {
+        Self::Generic(format!("processing error: {}", err))
+    }
 
     /// Unwound stracktrace of a panic.
-    #[error("panic: {0}")]
-    Panic(#[source] anyhow::Error),
+    pub fn panic(err: anyhow::Error) -> Self {
+        Self::Generic(format!("panic: {}", err))
+    }
 
     /// Error during actor initialization.
-    #[error("initialization error: {0}")]
-    Init(#[source] anyhow::Error),
+    pub fn init(err: anyhow::Error) -> Self {
+        Self::Generic(format!("initialization error: {}", err))
+    }
 
     /// An underlying mailbox error.
-    #[error(transparent)]
-    Mailbox(#[from] MailboxError),
+    pub fn mailbox(err: MailboxError) -> Self {
+        Self::Generic(err.to_string())
+    }
 
     /// An underlying mailbox sender error.
-    #[error(transparent)]
-    MailboxSender(#[from] MailboxSenderError),
+    pub fn mailbox_sender(err: MailboxSenderError) -> Self {
+        Self::Generic(err.to_string())
+    }
 
     /// An underlying checkpoint error.
-    #[error("checkpoint error: {0}")]
-    Checkpoint(#[source] CheckpointError),
+    pub fn checkpoint(err: CheckpointError) -> Self {
+        Self::Generic(format!("checkpoint error: {}", err))
+    }
 
     /// An underlying message log error.
-    #[error("message log error: {0}")]
-    MessageLog(#[source] MessageLogError),
+    pub fn message_log(err: MessageLogError) -> Self {
+        Self::Generic(format!("message log error: {}", err))
+    }
 
     /// The actor's state could not be determined.
-    #[error("actor is in an indeterminate state")]
-    IndeterminateState,
-
-    /// An actor supervision event was not handled.
-    #[error("supervision: {0}")]
-    UnhandledSupervisionEvent(#[from] ActorSupervisionEvent),
-
-    /// A special kind of error that allows us to clone errors: we can keep the
-    /// error string, but we lose the error structure.
-    #[error("{0}")]
-    Passthrough(#[from] anyhow::Error),
+    pub fn indeterminate_state() -> Self {
+        Self::Generic("actor is in an indeterminate state".to_string())
+    }
 }
 
 impl ActorError {
@@ -368,14 +377,6 @@ impl ActorError {
             kind: Box::new(kind),
         }
     }
-
-    /// Passthrough this error.
-    fn passthrough(&self) -> Self {
-        ActorError::new(
-            &self.actor_id,
-            ActorErrorKind::Passthrough(anyhow::anyhow!("{}", self.kind)),
-        )
-    }
 }
 
 impl fmt::Display for ActorError {
@@ -395,7 +396,7 @@ impl From<MailboxError> for ActorError {
     fn from(inner: MailboxError) -> Self {
         Self {
             actor_id: Box::new(inner.actor_id().clone()),
-            kind: Box::new(ActorErrorKind::from(inner)),
+            kind: Box::new(ActorErrorKind::mailbox(inner)),
         }
     }
 }
@@ -404,7 +405,7 @@ impl From<MailboxSenderError> for ActorError {
     fn from(inner: MailboxSenderError) -> Self {
         Self {
             actor_id: Box::new(inner.location().actor_id().clone()),
-            kind: Box::new(ActorErrorKind::from(inner)),
+            kind: Box::new(ActorErrorKind::mailbox_sender(inner)),
         }
     }
 }
@@ -413,7 +414,7 @@ impl From<ActorSupervisionEvent> for ActorError {
     fn from(inner: ActorSupervisionEvent) -> Self {
         Self {
             actor_id: Box::new(inner.actor_id.clone()),
-            kind: Box::new(ActorErrorKind::UnhandledSupervisionEvent(inner)),
+            kind: Box::new(ActorErrorKind::UnhandledSupervisionEvent(Box::new(inner))),
         }
     }
 }
@@ -473,9 +474,8 @@ pub enum ActorStatus {
     Stopping,
     /// The actor is stopped. It is no longer processing messages.
     Stopped,
-    /// The actor failed with the provided actor error formatted in string
-    /// representation.
-    Failed(String),
+    /// The actor failed with the provided actor error.
+    Failed(ActorErrorKind),
 }
 
 impl ActorStatus {
@@ -489,23 +489,11 @@ impl ActorStatus {
         matches!(self, Self::Failed(_))
     }
 
-    /// Create a passthrough of this status. The returned status is a clone,
-    /// except that [`ActorStatus::Failed`] is replaced with its passthrough.
-    fn passthrough(&self) -> Self {
-        match self {
-            Self::Unknown => Self::Unknown,
-            Self::Created => Self::Created,
-            Self::Initializing => Self::Initializing,
-            Self::Client => Self::Client,
-            Self::Idle => Self::Idle,
-            Self::Processing(instant, handler) => Self::Processing(*instant, handler.clone()),
-            Self::Saving(instant) => Self::Saving(*instant),
-            Self::Loading(instant) => Self::Loading(*instant),
-            Self::Stopping => Self::Stopping,
-            Self::Stopped => Self::Stopped,
-            Self::Failed(err) => Self::Failed(err.clone()),
-        }
+    /// 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()
     }
@@ -662,7 +650,7 @@ impl<A: Actor> IntoFuture for ActorHandle<A> {
             let result = status_receiver.wait_for(ActorStatus::is_terminal).await;
             match result {
                 Err(_) => ActorStatus::Unknown,
-                Ok(status) => status.passthrough(),
+                Ok(status) => status.clone(),
             }
         };
 

hyperactor/src/mailbox.rs

@@ -3142,8 +3142,8 @@ mod tests {
         let ActorStatus::Failed(ref msg) = *foo_status.borrow() else {
             unreachable!()
         };
-        assert!(msg.as_str().contains(
-            "serving quux[0].foo[0]: processing error: a message from \
+        assert!(msg.to_string().contains(
+            "processing error: a message from \
                 quux[0].foo[0] to corge[0].bar[0][9999] was undeliverable and returned"
         ));
 

hyperactor/src/mailbox/undeliverable.rs

@@ -165,7 +165,7 @@ pub fn supervise_undeliverable_messages_with<R, F>(
 
                     if let Err(e) = sink.send(ActorSupervisionEvent::new(
                         actor_id,
-                        ActorStatus::Failed(format!("message not delivered: {}", env)),
+                        ActorStatus::generic_failure(format!("message not delivered: {}", env)),
                         Some(headers),
                         None,
                     )) {

hyperactor/src/proc.rs

@@ -1102,18 +1102,21 @@ impl<A: Actor> Instance<A> {
         let (actor_status, event) = match result {
             Ok(_) => (ActorStatus::Stopped, None),
             Err(ActorError {
-                kind: box ActorErrorKind::UnhandledSupervisionEvent(event),
+                kind: box ActorErrorKind::UnhandledSupervisionEvent(box event),
                 ..
             }) => (event.actor_status.clone(), Some(event)),
-            Err(err) => (
-                ActorStatus::Failed(err.to_string()),
-                Some(ActorSupervisionEvent::new(
-                    self.cell.actor_id().clone(),
-                    ActorStatus::Failed(err.to_string()),
-                    None,
-                    None,
-                )),
-            ),
+            Err(err) => {
+                let error_kind = ActorErrorKind::Generic(err.kind.to_string());
+                (
+                    ActorStatus::Failed(error_kind.clone()),
+                    Some(ActorSupervisionEvent::new(
+                        self.cell.actor_id().clone(),
+                        ActorStatus::Failed(error_kind),
+                        None,
+                        None,
+                    )),
+                )
+            }
         };
 
         if let Some(parent) = self.cell.maybe_unlink_parent() {
@@ -1174,7 +1177,10 @@ impl<A: Actor> Instance<A> {
                     .unwrap_or_else(|e| format!("Cannot take backtrace due to: {:?}", e));
                 Err(ActorError::new(
                     self.self_id(),
-                    ActorErrorKind::Panic(anyhow::anyhow!("{}\n{}", err_msg, backtrace)),
+                    ActorErrorKind::panic(anyhow::anyhow!(
+                        "{}
+{}", err_msg, backtrace
+                    )),
                 ))
             }
         };
@@ -1244,7 +1250,7 @@ impl<A: Actor> Instance<A> {
         actor
             .init(self)
             .await
-            .map_err(|err| ActorError::new(self.self_id(), ActorErrorKind::Init(err)))?;
+            .map_err(|err| ActorError::new(self.self_id(), ActorErrorKind::init(err)))?;
         let need_drain;
         'messages: loop {
             self.change_status(ActorStatus::Idle);
@@ -1260,7 +1266,7 @@ impl<A: Actor> Instance<A> {
                         for supervision_event in supervision_event_receiver.drain() {
                             self.handle_supervision_event(actor, supervision_event).await?;
                         }
-                        return Err(ActorError::new(self.self_id(), ActorErrorKind::Processing(err)));
+                        return Err(ActorError::new(self.self_id(), ActorErrorKind::processing(err)));
                     }
                 }
                 signal = signal_receiver.recv() => {
@@ -1293,7 +1299,7 @@ impl<A: Actor> Instance<A> {
                 if let Err(err) = work.handle(actor, self).await {
                     return Err(ActorError::new(
                         self.self_id(),
-                        ActorErrorKind::Processing(err),
+                        ActorErrorKind::processing(err),
                     ));
                 }
                 n += 1;
@@ -1323,7 +1329,7 @@ impl<A: Actor> Instance<A> {
                 // The supervision event wasn't handled by this actor, chain it and bubble it up.
                 let supervision_event = ActorSupervisionEvent::new(
                     self.self_id().clone(),
-                    ActorStatus::Failed("did not handle supervision event".to_string()),
+                    ActorStatus::generic_failure("did not handle supervision event"),
                     None,
                     Some(Box::new(supervision_event)),
                 );
@@ -1334,7 +1340,10 @@ impl<A: Actor> Instance<A> {
                 // Create a new supervision event for this failure and propagate it.
                 let supervision_event = ActorSupervisionEvent::new(
                     self.self_id().clone(),
-                    ActorStatus::Failed(format!("failed to handle supervision event: {}", err)),
+                    ActorStatus::generic_failure(format!(
+                        "failed to handle supervision event: {}",
+                        err
+                    )),
                     None,
                     Some(Box::new(supervision_event)),
                 );
@@ -2388,17 +2397,17 @@ mod tests {
                 "some random failure"
             )))
             .unwrap();
-        let root_2_actor_id = root_2.actor_id().clone();
+        let _root_2_actor_id = root_2.actor_id().clone();
         assert_matches!(
             root_2.await,
-            ActorStatus::Failed(err) if err == format!("serving {}: processing error: some random failure", root_2_actor_id)
+            ActorStatus::Failed(err) if err.to_string() == "processing error: some random failure"
         );
 
         // TODO: should we provide finer-grained stop reasons, e.g., to indicate it was
         // stopped by a parent failure?
-        assert_eq!(
+        assert_matches!(
             root.await,
-            ActorStatus::Failed("did not handle supervision event".to_string())
+            ActorStatus::Failed(ActorErrorKind::Generic(msg)) if msg == "did not handle supervision event"
         );
         assert_eq!(root_2_1.await, ActorStatus::Stopped);
         assert_eq!(root_1.await, ActorStatus::Stopped);
@@ -2913,16 +2922,13 @@ mod tests {
             // The time field is ignored for Eq and PartialEq.
             ActorSupervisionEvent::new(
                 child_actor_id,
-                ActorStatus::Failed(
+                ActorStatus::generic_failure(
                     "failed to handle supervision event: failed to handle supervision event!"
-                        .to_string(),
                 ),
                 None,
                 Some(Box::new(ActorSupervisionEvent::new(
                     grandchild_actor_id,
-                    ActorStatus::Failed(
-                        "serving local[0].parent[2]: processing error: trigger failure".to_string()
-                    ),
+                    ActorStatus::generic_failure("processing error: trigger failure"),
                     None,
                     None,
                 ))),

hyperactor/src/supervision.rs

@@ -22,6 +22,7 @@ use serde::Serialize;
 
 use crate as hyperactor; // for macros
 use crate::Named;
+use crate::actor::ActorErrorKind;
 use crate::actor::ActorStatus;
 use crate::attrs::Attrs;
 use crate::reference::ActorId;
@@ -63,10 +64,12 @@ impl ActorSupervisionEvent {
     /// Compute an actor status from this event, ensuring that "caused-by"
     /// events are included in failure states. This should be used as the
     /// actor status when reporting events to users.
-    pub fn status(&self) -> ActorStatus {
-        match &self.actor_status {
-            ActorStatus::Failed(msg) => ActorStatus::Failed(format!("{}: {}", self, msg)),
-            status => status.clone(),
+    pub fn status(self) -> ActorStatus {
+        match self.actor_status {
+            ActorStatus::Failed(_) => {
+                ActorStatus::Failed(ActorErrorKind::UnhandledSupervisionEvent(Box::new(self)))
+            }
+            status => status,
         }
     }
 }

hyperactor_mesh/src/proc_mesh.rs

@@ -842,7 +842,7 @@ impl ProcEvents {
                         // TODO(T231868026): find a better way to represent all actors in a proc for supervision event
                         let event = ActorSupervisionEvent::new(
                             proc_id.actor_id("any", 0),
-                            ActorStatus::Failed(format!("proc {} is stopped", proc_id)),
+                            ActorStatus::generic_failure(format!("proc {} is stopped", proc_id)),
                             None,
                             None,
                         );