[hyperactor]: mailbox: CanSend for Mailbox use mailbox port for returns

Pull Request resolved: #303

 the `CanSend` for `Mailbox` produces a self-bound `PortHandle` (allocating and binding only if needed). i don't have any immediate test coverage but removing `monitored_return_handle()` from this code is a definite and substantial improvement.
ghstack-source-id: 291349747

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

**NOTE FOR REVIEWERS**: This PR has internal Meta-specific changes or comments, please review them on [Phabricator](https://our.internmc.facebook.com/intern/diff/D76926674/)!

Author: Shayne Fletcher

hyperactor/src/mailbox/mod.rs

@@ -1082,6 +1082,33 @@ impl Mailbox {
         MailboxError::new(self.state.actor_id.clone(), err)
     }
 
+    /// Look up the `UnboundedPortSender` for the port associated with
+    /// the message type `M`, as defined by `<M as Named>::port()`.
+    ///
+    /// This performs a dynamic downcast to recover the original
+    /// sender type. Returns `None` if the port is not bound or if the
+    /// type does not match.
+    ///
+    /// # Panics
+    /// Panics in debug mode if the port is bound but the stored
+    /// `port_id` does not match the expected one.
+    pub(crate) fn lookup_sender<M: RemoteMessage>(&self) -> Option<UnboundedPortSender<M>> {
+        let port_index = M::port();
+        self.state.ports.get(&port_index).and_then(|boxed| {
+            boxed
+                .as_any()
+                .downcast_ref::<UnboundedSender<M>>()
+                .map(|s| {
+                    debug_assert_eq!(
+                        s.port_id,
+                        self.actor_id().port_id(port_index),
+                        "port_id mismatch in downcasted UnboundedSender"
+                    );
+                    s.sender.clone()
+                })
+        })
+    }
+
     fn bind<M: RemoteMessage>(&self, handle: &PortHandle<M>) -> PortRef<M> {
         assert_eq!(
             handle.mailbox.actor_id(),
@@ -1210,8 +1237,17 @@ impl MailboxSender for Mailbox {
 
 impl cap::sealed::CanSend for Mailbox {
     fn post(&self, dest: PortId, data: Serialized) {
+        let undeliverable_port_id = Undeliverable::<MessageEnvelope>::port();
+        let return_handle = match self.lookup_sender::<Undeliverable<MessageEnvelope>>() {
+            Some(sender) => PortHandle::new(self.clone(), undeliverable_port_id, sender),
+            None => {
+                let (handle, _) = self.open_port::<Undeliverable<MessageEnvelope>>();
+                handle.bind_to(undeliverable_port_id);
+                handle
+            }
+        };
         let envelope = MessageEnvelope::new(self.actor_id().clone(), dest, data);
-        MailboxSender::post(self, envelope, monitored_return_handle());
+        MailboxSender::post(self, envelope, return_handle);
     }
 }
 
@@ -1592,6 +1628,8 @@ pub struct SerializedSenderError {
 ///   - It abstracts over [`Port`]s and [`OncePort`]s, by dynamically tracking the
 ///     validity of the underlying port.
 trait SerializedSender: Send + Sync {
+    fn as_any(&self) -> &dyn Any;
+
     /// Send a serialized message. SerializedSender will deserialize the
     /// message (failing if it fails to deserialize), and then send the
     /// resulting message on the underlying port.
@@ -1603,7 +1641,7 @@ trait SerializedSender: Send + Sync {
 }
 
 /// A sender to an M-typed unbounded port.
-enum UnboundedPortSender<M: Message> {
+pub(crate) enum UnboundedPortSender<M: Message> {
     /// Send directly to the mpsc queue.
     Mpsc(mpsc::UnboundedSender<M>),
     /// Use the provided function to enqueue the item.
@@ -1675,6 +1713,10 @@ impl<M: Message> Clone for UnboundedSender<M> {
 }
 
 impl<M: RemoteMessage> SerializedSender for UnboundedSender<M> {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     fn send_serialized(&self, serialized: Serialized) -> Result<bool, SerializedSenderError> {
         match serialized.deserialized() {
             Ok(message) => {
@@ -1757,6 +1799,10 @@ impl<M: Message> Clone for OnceSender<M> {
 }
 
 impl<M: RemoteMessage> SerializedSender for OnceSender<M> {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     fn send_serialized(&self, serialized: Serialized) -> Result<bool, SerializedSenderError> {
         match serialized.deserialized() {
             Ok(message) => self.send_once(message).map_err(|e| SerializedSenderError {
@@ -1781,6 +1827,10 @@ struct UntypedUnboundedSender {
 }
 
 impl SerializedSender for UntypedUnboundedSender {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     fn send_serialized(&self, serialized: Serialized) -> Result<bool, SerializedSenderError> {
         (self.sender)(serialized).map_err(|(data, err)| SerializedSenderError {
             data,