Ignore mailbox errors in GetState for HostMeshAgent and ProcMeshAgent (#1691)

Summary:
Pull Request resolved: #1691

If some actor owns some ActorMeshes, it will periodically send out a GetState message
to the HostMeshAgent and ProcMeshAgent. If that sender crashes while waiting for a reply,
it'll cause a MailboxSenderError on the agents.
We don't want those agents to stop because of such an error, as it just means nobody will
receive a reply.

GetRankStatus and GetState messages are read-only and have no side effects, so it's fine
to just warn on the MailboxSenderError, there is no invalid state left hanging around.

Reviewed By: vidhyav, xunnanxu

Differential Revision: D85720817

fbshipit-source-id: 4e2881553590b08e60ca7a264eb669fa7ff082ec

Author: dulinriley

hyperactor_mesh/src/proc_mesh/mesh_agent.rs

@@ -669,8 +669,18 @@ impl Handler<resource::GetRankStatus> for ProcMeshAgent {
             StatusOverlay::try_from_runs(vec![(rank..(rank + 1), status)])
                 .expect("valid single-run overlay")
         };
-        get_rank_status.reply.send(cx, overlay)?;
-
+        let result = get_rank_status.reply.send(cx, overlay);
+        // Ignore errors, because returning Err from here would cause the ProcMeshAgent
+        // to be stopped, which would prevent querying and spawning other actors.
+        // This only means some actor that requested the state of an actor failed to receive it.
+        if let Err(e) = result {
+            tracing::warn!(
+                actor = %cx.self_id(),
+                "failed to send GetRankStatus reply to {} due to error: {}",
+                get_rank_status.reply.port_id().actor_id(),
+                e
+            );
+        }
         Ok(())
     }
 }
@@ -724,7 +734,18 @@ impl Handler<resource::GetState<ActorState>> for ProcMeshAgent {
             },
         };
 
-        get_state.reply.send(cx, state)?;
+        let result = get_state.reply.send(cx, state);
+        // Ignore errors, because returning Err from here would cause the ProcMeshAgent
+        // to be stopped, which would prevent querying and spawning other actors.
+        // This only means some actor that requested the state of an actor failed to receive it.
+        if let Err(e) = result {
+            tracing::warn!(
+                actor = %cx.self_id(),
+                "failed to send GetState reply to {} due to error: {}",
+                get_state.reply.port_id().actor_id(),
+                e
+            );
+        }
         Ok(())
     }
 }

hyperactor_mesh/src/v1/host_mesh/mesh_agent.rs

@@ -290,7 +290,18 @@ impl Handler<resource::GetRankStatus> for HostMeshAgent {
             StatusOverlay::try_from_runs(vec![(rank..(rank + 1), status)])
                 .expect("valid single-run overlay")
         };
-        get_rank_status.reply.send(cx, overlay)?;
+        let result = get_rank_status.reply.send(cx, overlay);
+        // Ignore errors, because returning Err from here would cause the HostMeshAgent
+        // to be stopped, which would take down the entire host. This only means
+        // some actor that requested the rank status failed to receive it.
+        if let Err(e) = result {
+            tracing::warn!(
+                actor = %cx.self_id(),
+                "failed to send GetRankStatus reply to {} due to error: {}",
+                get_rank_status.reply.port_id().actor_id(),
+                e
+            );
+        }
         Ok(())
     }
 }
@@ -403,7 +414,18 @@ impl Handler<resource::GetState<ProcState>> for HostMeshAgent {
             },
         };
 
-        get_state.reply.send(cx, state)?;
+        let result = get_state.reply.send(cx, state);
+        // Ignore errors, because returning Err from here would cause the HostMeshAgent
+        // to be stopped, which would take down the entire host. This only means
+        // some actor that requested the state of a proc failed to receive it.
+        if let Err(e) = result {
+            tracing::warn!(
+                actor = %cx.self_id(),
+                "failed to send GetState reply to {} due to error: {}",
+                get_state.reply.port_id().actor_id(),
+                e
+            );
+        }
         Ok(())
     }
 }