: clock: bind undeliverable port in sim clock's mailbox

Summary:
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 if an undeliverable message is encountered in this context.

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

Differential Revision: D78191921

Author: shayne-fletcher

hyperactor/src/clock.rs

@@ -12,16 +12,21 @@ 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::MessageEnvelope;
+use crate::mailbox::Undeliverable;
 use crate::simnet::SleepEvent;
 use crate::simnet::simnet_handle;
 
@@ -175,6 +180,30 @@ impl ClockKind {
     }
 }
 
+// TODO (SF, 2025-07-11): Remove this global, thread through a mailbox
+// from upstack and handle undeliverable messages properly.
+static SIMCLOCK_MAILBOX: OnceLock<Mailbox> = OnceLock::new();
+fn simclock_mailbox_id() -> ActorId {
+    id!(proc[0].proc).clone()
+}
+fn simclock_mailbox() -> &'static Mailbox {
+    SIMCLOCK_MAILBOX.get_or_init(|| {
+        let mailbox = Mailbox::new_detached(simclock_mailbox_id());
+        let (undeliverable_messages, mut rx) =
+            mailbox.open_port::<Undeliverable<MessageEnvelope>>();
+        undeliverable_messages.bind_to(Undeliverable::<MessageEnvelope>::port());
+        tokio::spawn(async move {
+            if let Ok(Undeliverable(mut envelope)) = rx.recv().await {
+                tracing::error!(
+                    "simclock mailbox got an undeliverable message; calling `abort()`!"
+                );
+                std::process::abort();
+            }
+        });
+        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 +213,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 +228,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 +263,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

@@ -1310,7 +1310,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,