clock: bind undeliverable port in sim clock's mailbox (#517)

Summary:
Pull Request resolved: #517

(1) pesky warnings in python about using a mailbox for sending without binding the undeliverable port (see D78031232) turned out to be due to the use of `SimClock`.

hack in a "fix" that ~~terminates the process~~ logs the event (the idea of terminating the process didn't work out - test failures[*] ) if an undeliverable message is encountered in this context.

we'll definitely want to revisit this however (kaiyuan-li, thomasywang) and fix it more appropriately (maybe by threading through a mailbox from upstack that has the undeliverable port bound and bounces undeliverables into supervision events perhaps?).

(2) provide `unused_return_handle()` and use it to replace the `/*unused*/ monitored_return_handle()` idiom from the codebase.

 ---

[*] the failed tests being,

(1) `buck test 'fbcode//mode/dev-nosan' fbcode//monarch/controller:controller-unittest -- --exact 'monarch/controller:controller-unittest - tests::test_sim_supervision_failure'`,

(2)`buck test 'fbcode//mode/opt' fbcode//monarch/python/tests:test_sim_backend -- --exact 'monarch/python/tests:test_sim_backend - test_sim_backend.py::TestSimBackend::test_local_mesh_setup'`,

(3) `buck test 'fbcode//mode/dev-nosan' fbcode//monarch/hyperactor:hyperactor-unittest -- --exact 'monarch/hyperactor:hyperactor-unittest - clock::tests::test_sim_timeout'`

in all cases i *think* the tests themselves are passing and the undeliverable messages are actually encountered during teardown.

Differential Revision: D78191921

Author: shayne-fletcher

hyperactor/src/clock.rs

@@ -12,16 +12,25 @@ use std::error::Error;
 use std::fmt;
 use std::sync::LazyLock;
 use std::sync::Mutex;
+use std::sync::OnceLock;
 use std::time::SystemTime;
 
 use futures::pin_mut;
 use hyperactor_telemetry::TelemetryClock;
 use serde::Deserialize;
 use serde::Serialize;
 
+use crate::ActorId;
 use crate::Mailbox;
 use crate::channel::ChannelAddr;
+use crate::data::Named;
 use crate::id;
+use crate::mailbox::DeliveryError;
+use crate::mailbox::MailboxSender;
+use crate::mailbox::MessageEnvelope;
+use crate::mailbox::Undeliverable;
+use crate::mailbox::UndeliverableMailboxSender;
+use crate::mailbox::unused_return_handle;
 use crate::simnet::SleepEvent;
 use crate::simnet::simnet_handle;
 
@@ -175,6 +184,27 @@ impl ClockKind {
     }
 }
 
+// TODO (SF, 2025-07-11): Remove this global, thread through a mailbox
+// from upstack and handle undeliverable messages properly.
+fn simclock_mailbox() -> &'static Mailbox {
+    static SIMCLOCK_MAILBOX: OnceLock<Mailbox> = OnceLock::new();
+    SIMCLOCK_MAILBOX.get_or_init(|| {
+        let mailbox = Mailbox::new_detached(id!(proc[0].proc).clone());
+        let (undeliverable_messages, mut rx) =
+            mailbox.open_port::<Undeliverable<MessageEnvelope>>();
+        undeliverable_messages.bind_to(Undeliverable::<MessageEnvelope>::port());
+        tokio::spawn(async move {
+            while let Ok(Undeliverable(mut envelope)) = rx.recv().await {
+                envelope.try_set_error(DeliveryError::BrokenLink(
+                    "message returned to undeliverable port".to_string(),
+                ));
+                UndeliverableMailboxSender.post(envelope, unused_return_handle());
+            }
+        });
+        mailbox
+    })
+}
+
 /// Clock to be used in simulator runs that allows the simnet to create a scheduled event for.
 /// When the wakeup event becomes the next earliest scheduled event, the simnet will advance it's
 /// time to the wakeup time and use the transmitter to wake up this green thread
@@ -184,7 +214,7 @@ pub struct SimClock;
 impl Clock for SimClock {
     /// Tell the simnet to wake up this green thread after the specified duration has pass on the simnet
     async fn sleep(&self, duration: tokio::time::Duration) {
-        let mailbox = Mailbox::new_detached(id!(proc[0].proc).clone());
+        let mailbox = simclock_mailbox().clone();
         let (tx, rx) = mailbox.open_once_port::<()>();
 
         simnet_handle()
@@ -199,7 +229,7 @@ impl Clock for SimClock {
     }
 
     async fn non_advancing_sleep(&self, duration: tokio::time::Duration) {
-        let mailbox = Mailbox::new_detached(id!(proc[0].proc).clone());
+        let mailbox = simclock_mailbox().clone();
         let (tx, rx) = mailbox.open_once_port::<()>();
 
         simnet_handle()
@@ -234,7 +264,7 @@ impl Clock for SimClock {
     where
         F: std::future::Future<Output = T>,
     {
-        let mailbox = Mailbox::new_detached(id!(proc[0].proc).clone());
+        let mailbox = simclock_mailbox().clone();
         let (tx, deadline_rx) = mailbox.open_once_port::<()>();
 
         simnet_handle()

hyperactor/src/mailbox.rs

@@ -128,6 +128,7 @@ pub use undeliverable::Undeliverable;
 pub use undeliverable::UndeliverableMessageError;
 pub use undeliverable::monitored_return_handle; // TODO: Audit
 pub use undeliverable::supervise_undeliverable_messages;
+pub use undeliverable::unused_return_handle;
 /// For [`MailboxAdminMessage`], a message type for mailbox administration.
 pub mod mailbox_admin_message;
 pub use mailbox_admin_message::MailboxAdminMessage;
@@ -1310,7 +1311,7 @@ impl cap::sealed::CanSend for Mailbox {
                 .get_or_init(DashSet::new)
                 .insert(actor_id.clone())
             {
-                let bt = std::backtrace::Backtrace::capture();
+                let bt = std::backtrace::Backtrace::force_capture();
                 tracing::warn!(
                     actor_id = ?actor_id,
                     backtrace = ?bt,
@@ -2833,8 +2834,7 @@ mod tests {
                 envelope.try_set_error(DeliveryError::BrokenLink(
                     "returned in unit test".to_string(),
                 ));
-                UndeliverableMailboxSender
-                    .post(envelope, /*unused */ monitored_return_handle().clone());
+                UndeliverableMailboxSender.post(envelope, unused_return_handle().clone());
             }
         });
         drop(return_handle);

hyperactor/src/mailbox/undeliverable.rs

@@ -53,15 +53,12 @@ static MONITORED_RETURN_HANDLE: OnceLock<PortHandle<Undeliverable<MessageEnvelop
 pub fn monitored_return_handle() -> PortHandle<Undeliverable<MessageEnvelope>> {
     let return_handle = MONITORED_RETURN_HANDLE.get_or_init(|| {
         let (return_handle, mut rx) = new_undeliverable_port();
-        // Don't reuse `return_handle` for `h`: else it will never get
-        // dropped and the task will never return.
-        let (h, _) = new_undeliverable_port();
         crate::init::get_runtime().spawn(async move {
             while let Ok(Undeliverable(mut envelope)) = rx.recv().await {
                 envelope.try_set_error(DeliveryError::BrokenLink(
                     "message returned to undeliverable port".to_string(),
                 ));
-                super::UndeliverableMailboxSender.post(envelope, /*unused */ h.clone());
+                super::UndeliverableMailboxSender.post(envelope, unused_return_handle());
             }
         });
         return_handle
@@ -70,14 +67,19 @@ pub fn monitored_return_handle() -> PortHandle<Undeliverable<MessageEnvelope>> {
     return_handle.clone()
 }
 
+/// For when a return handle is required but we know it won't be used.
+pub fn unused_return_handle() -> PortHandle<Undeliverable<MessageEnvelope>> {
+    panic!("unused_return_handle was called: this indicates a logic error; the return handle was unexpectedly used")
+}
+
 /// Returns a message envelope to its original sender.
 pub(crate) fn return_undeliverable(
     return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
     envelope: MessageEnvelope,
 ) {
     let envelope_copy = envelope.clone();
     if (return_handle.send(Undeliverable(envelope))).is_err() {
-        UndeliverableMailboxSender.post(envelope_copy, /*unsued*/ return_handle)
+        UndeliverableMailboxSender.post(envelope_copy, unused_return_handle())
     }
 }
 
@@ -149,8 +151,7 @@ pub fn supervise_undeliverable_messages(
                 ))
                 .is_err()
             {
-                UndeliverableMailboxSender
-                    .post(envelope.clone(), /*unused*/ monitored_return_handle())
+                UndeliverableMailboxSender.post(envelope.clone(), unused_return_handle())
             }
         }
     });

hyperactor_mesh/src/comm.rs

@@ -31,7 +31,7 @@ use hyperactor::mailbox::MailboxSender;
 use hyperactor::mailbox::Undeliverable;
 use hyperactor::mailbox::UndeliverableMailboxSender;
 use hyperactor::mailbox::UndeliverableMessageError;
-use hyperactor::mailbox::monitored_return_handle;
+use hyperactor::mailbox::unused_return_handle;
 use hyperactor::reference::UnboundPort;
 use ndslice::selection::routing::RoutingFrame;
 use serde::Deserialize;
@@ -203,8 +203,7 @@ impl Actor for CommActor {
         }
 
         // 3. A return of an undeliverable message was itself returned.
-        UndeliverableMailboxSender
-            .post(message_envelope, /*unused */ monitored_return_handle());
+        UndeliverableMailboxSender.post(message_envelope, unused_return_handle());
         Ok(())
     }
 }