: more baseline behavior testing

Summary:
this diff adds another test to probe behaviors in the event of message send failures. the approach taken here is to deliberately target sending messages to actors (or non-existent) actors directly. i'm interested in feedback if the approaches taken here are meaningful or valid.
ghstack-source-id: 288208548

Differential Revision: D75975772

Author: shayne-fletcher

hyperactor_mesh/examples/dining_philosophers.rs

@@ -21,7 +21,6 @@ use hyperactor::PortRef;
 use hyperactor::message::IndexedErasedUnbound;
 use hyperactor_mesh::Mesh;
 use hyperactor_mesh::ProcMesh;
-use hyperactor_mesh::RootActorMesh;
 use hyperactor_mesh::actor_mesh::ActorMesh;
 use hyperactor_mesh::actor_mesh::Cast;
 use hyperactor_mesh::alloc::AllocSpec;

hyperactor_mesh/src/actor_mesh.rs

@@ -316,6 +316,7 @@ pub(crate) mod test_util {
         Cast<(String, PortRef<String>)>,
         Cast<GetRank>,
         Cast<Error>,
+        GetRank,
         Relay,
         IndexedErasedUnbound<Cast<(String, PortRef<String>)>>,
         IndexedErasedUnbound<Cast<GetRank>>,
@@ -362,6 +363,20 @@ pub(crate) mod test_util {
         }
     }
 
+    #[async_trait]
+    impl Handler<GetRank> for TestActor {
+        async fn handle(
+            &mut self,
+            this: &Instance<Self>,
+            GetRank(ok, reply): GetRank,
+        ) -> Result<(), anyhow::Error> {
+            let rank = this.self_id().rank();
+            reply.send(this, rank)?;
+            anyhow::ensure!(ok, "intentional error!"); // If `!ok` exit with `Err()`.
+            Ok(())
+        }
+    }
+
     #[async_trait]
     impl Handler<Cast<(String, PortRef<String>)>> for TestActor {
         async fn handle(
@@ -662,6 +677,60 @@ mod tests {
             );
             assert!(events.next().await.is_none());
         }
+
+        #[tracing_test::traced_test]
+        #[tokio::test]
+        async fn test_behaviors_on_failed_send() {
+            use hyperactor::ActorId;
+            use hyperactor::ProcId;
+            use hyperactor::WorldId;
+
+            let alloc = LocalAllocator
+                .allocate(AllocSpec {
+                    shape: shape! { replica = 1  },
+                    constraints: Default::default(),
+                })
+                .await
+                .unwrap();
+
+            let stop = alloc.stopper();
+            let name = alloc.name().to_string();
+            let mesh = ProcMesh::allocate(alloc).await.unwrap();
+
+            let actor_mesh = mesh.spawn::<TestActor>("foo", &()).await.unwrap();
+
+            let (reply_handle, mut reply_receiver) = actor_mesh.open_port();
+
+            // Send a message to an actor that exists.
+            let foo: ActorRef<TestActor> =
+                ActorRef::attest(ActorId(ProcId(WorldId(name.clone()), 0), "foo".into(), 0));
+            foo.send(mesh.client(), GetRank(true, reply_handle.bind()))
+                .unwrap();
+            // This is ok.
+            let rank = reply_receiver.recv().await.unwrap();
+            assert_eq!(rank, 0);
+
+            // Send a message to an actor that doesn't exist but the proc does.
+            let _bar: ActorRef<TestActor> =
+                ActorRef::attest(ActorId(ProcId(WorldId(name.clone()), 0), "foo".into(), 1));
+            // Uncomment for infinite hang.
+            /*
+            // Message gets logged.
+            // delivery error: address not routable: no mailbox for actor _1ku59VQkwPSM[0].foo[1] registered in muxer
+            bar.send(mesh.client(), GetRank(true, reply_handle.bind())).unwrap();
+            */
+
+            // Send a message to an actor on a proc that doesn't exist.
+            let _baz: ActorRef<TestActor> =
+                ActorRef::attest(ActorId(ProcId(WorldId(name.clone()), 1), "foo".into(), 0));
+            // Uncomment for stack overflow.
+            /*
+            baz.send(mesh.client(), GetRank(true, reply_handle.bind())).unwrap();
+            */
+
+            // Stop the mesh.
+            stop();
+        }
     } // mod local
 
     mod process {