Reformat ActorSupervisionEvent

Differential Revision: [D86925582](https://our.internmc.facebook.com/intern/diff/D86925582/)

ghstack-source-id: 323132637
Pull Request resolved: #1881

Author: zdevito

hyperactor/src/actor.rs

@@ -131,6 +131,13 @@ pub trait Actor: Sized + Send + Debug + 'static {
     ) -> 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
@@ -321,16 +328,24 @@ pub enum ActorErrorKind {
     #[error("{0}")]
     Generic(String),
 
-    /// An actor supervision event was not handled.
-    #[error("supervision: {0}")]
+    /// 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 {
-        Self::Generic(format!("processing error: {}", err))
+        // 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.
@@ -410,15 +425,6 @@ impl From<MailboxSenderError> for ActorError {
     }
 }
 
-impl From<ActorSupervisionEvent> for ActorError {
-    fn from(inner: ActorSupervisionEvent) -> Self {
-        Self {
-            actor_id: Box::new(inner.actor_id.clone()),
-            kind: Box::new(ActorErrorKind::UnhandledSupervisionEvent(Box::new(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.

hyperactor/src/mailbox/undeliverable.rs

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

hyperactor/src/proc.rs

@@ -1118,7 +1118,6 @@ impl<A: Actor> Instance<A> {
         let result = self
             .run_actor_tree(&mut actor, actor_loop_receivers, &mut work_rx)
             .await;
-
         let (actor_status, event) = match result {
             Ok(_) => (ActorStatus::Stopped, None),
             Err(ActorError {
@@ -1131,9 +1130,9 @@ impl<A: Actor> Instance<A> {
                     ActorStatus::Failed(error_kind.clone()),
                     Some(ActorSupervisionEvent::new(
                         self.cell.actor_id().clone(),
+                        actor.display_name(),
                         ActorStatus::Failed(error_kind),
                         None,
-                        None,
                     )),
                 )
             }
@@ -1286,7 +1285,11 @@ 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)));
+                        let kind = ActorErrorKind::processing(err);
+                        return Err(ActorError {
+                            actor_id: Box::new(self.self_id().clone()),
+                            kind: Box::new(kind),
+                        });
                     }
                 }
                 signal = signal_receiver.recv() => {
@@ -1346,28 +1349,17 @@ impl<A: Actor> Instance<A> {
                 Ok(())
             }
             Ok(false) => {
-                // 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::generic_failure("did not handle supervision event"),
-                    None,
-                    Some(Box::new(supervision_event)),
-                );
-                Err(supervision_event.into())
+                let kind = ActorErrorKind::UnhandledSupervisionEvent(Box::new(supervision_event));
+                Err(ActorError::new(self.self_id(), kind))
             }
             Err(err) => {
                 // The actor failed to handle the supervision event, it should die.
                 // Create a new supervision event for this failure and propagate it.
-                let supervision_event = ActorSupervisionEvent::new(
-                    self.self_id().clone(),
-                    ActorStatus::generic_failure(format!(
-                        "failed to handle supervision event: {}",
-                        err
-                    )),
-                    None,
-                    Some(Box::new(supervision_event)),
+                let kind = ActorErrorKind::ErrorDuringHandlingSupervision(
+                    err.to_string(),
+                    Box::new(supervision_event),
                 );
-                Err(supervision_event.into())
+                Err(ActorError::new(self.self_id(), kind))
             }
         }
     }
@@ -2857,107 +2849,6 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn test_supervision_event_handler_propagates() {
-        #[derive(Debug)]
-        struct FailingSupervisionActor;
-
-        #[async_trait]
-        impl Actor for FailingSupervisionActor {
-            type Params = ();
-
-            async fn new(_: ()) -> Result<Self, anyhow::Error> {
-                Ok(Self)
-            }
-
-            async fn handle_supervision_event(
-                &mut self,
-                _this: &Instance<Self>,
-                _event: &ActorSupervisionEvent,
-            ) -> Result<bool, anyhow::Error> {
-                anyhow::bail!("failed to handle supervision event!")
-            }
-        }
-
-        #[async_trait]
-        impl Handler<String> for FailingSupervisionActor {
-            async fn handle(
-                &mut self,
-                _cx: &crate::Context<Self>,
-                message: String,
-            ) -> anyhow::Result<()> {
-                Err(anyhow::anyhow!(message))
-            }
-        }
-
-        #[derive(Debug)]
-        struct ParentActor(tokio::sync::mpsc::UnboundedSender<ActorSupervisionEvent>);
-
-        #[async_trait]
-        impl Actor for ParentActor {
-            type Params = tokio::sync::mpsc::UnboundedSender<ActorSupervisionEvent>;
-
-            async fn new(
-                supervision_events: tokio::sync::mpsc::UnboundedSender<ActorSupervisionEvent>,
-            ) -> Result<Self, anyhow::Error> {
-                Ok(Self(supervision_events))
-            }
-
-            async fn handle_supervision_event(
-                &mut self,
-                _this: &Instance<Self>,
-                event: &ActorSupervisionEvent,
-            ) -> Result<bool, anyhow::Error> {
-                self.0.send(event.clone()).unwrap();
-                Ok(true)
-            }
-        }
-
-        let proc = Proc::local();
-
-        let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
-
-        let parent = proc.spawn::<ParentActor>("parent", event_tx).await.unwrap();
-        let child = proc
-            .spawn_child::<FailingSupervisionActor>(parent.cell().clone(), ())
-            .await
-            .unwrap();
-        let grandchild = proc
-            .spawn_child::<FailingSupervisionActor>(child.cell().clone(), ())
-            .await
-            .unwrap();
-
-        let child_actor_id = child.actor_id().clone();
-        let grandchild_actor_id = grandchild.actor_id().clone();
-
-        // Grandchild fails, triggering failure up the tree, finally receiving
-        // the event at the root.
-        grandchild.send("trigger failure".to_string()).unwrap();
-
-        assert!(grandchild.await.is_failed());
-        assert!(child.await.is_failed());
-
-        assert_eq!(
-            event_rx.recv().await.unwrap(),
-            // The time field is ignored for Eq and PartialEq.
-            ActorSupervisionEvent::new(
-                child_actor_id,
-                ActorStatus::generic_failure(
-                    "failed to handle supervision event: failed to handle supervision event!"
-                ),
-                None,
-                Some(Box::new(ActorSupervisionEvent::new(
-                    grandchild_actor_id,
-                    ActorStatus::generic_failure("processing error: trigger failure"),
-                    None,
-                    None,
-                ))),
-            )
-        );
-
-        assert!(event_rx.try_recv().is_err());
-    }
-
     #[tokio::test]
     async fn test_instance() {
         #[derive(Debug, Default, Actor)]