Derive actor telemetry IDs from ActorId (#4121)

Summary:

Actor telemetry IDs should use runtime actor identity, not routable actor addresses (changed in D102822047) This updates actor rows, sender IDs, receiver IDs, and actor status IDs to hash `ActorId` while keeping `ActorAddr` strings only for display/debug fields. This prevents location changes from changing the telemetry actor join key.

Reviewed By: mariusae

Differential Revision: D107156872

Author: thedavekwon

hyperactor/src/mailbox.rs

@@ -2163,15 +2163,15 @@ impl MailboxSender for Mailbox {
             return_undeliverable,
         } = metadata;
 
-        let to_actor_id = hash_to_u64(&dest);
+        let to_actor_id = hash_to_u64(dest.actor_addr().id());
         let message_id = hyperactor_telemetry::generate_message_id(to_actor_id);
         headers.set(crate::mailbox::headers::TELEMETRY_MESSAGE_ID, message_id);
         // Only set sender hash if not already present (cast path
         // pre-sets it with the originating actor).
         if !headers.contains_key(crate::mailbox::headers::SENDER_ACTOR_ID_HASH) {
             headers.set(
                 crate::mailbox::headers::SENDER_ACTOR_ID_HASH,
-                hash_to_u64(&sender),
+                hash_to_u64(sender.id()),
             );
         }
         headers.set(crate::mailbox::headers::TELEMETRY_PORT_ID, dest.index());

hyperactor/src/mailbox/headers.rs

@@ -32,7 +32,7 @@ declare_attrs! {
     /// The rust type of the message.
     pub attr RUST_MESSAGE_TYPE: String;
 
-    /// Hashed ActorAddr of the message sender, injected in post_unchecked().
+    /// Hashed ActorId of the message sender, injected in post_unchecked().
     pub attr SENDER_ACTOR_ID_HASH: u64;
 
     /// Full ActorAddr of the session owner — the actor whose Sequencer

hyperactor/src/proc.rs

@@ -2345,7 +2345,7 @@ impl<A: Actor> Instance<A> {
                 caller = %PanicLocation::caller(),
                 change_reason,
             );
-            let actor_id = hash_to_u64(self.self_addr());
+            let actor_id = hash_to_u64(self.self_addr().id());
             notify_actor_status_changed(ActorStatusEvent {
                 id: generate_actor_status_event_id(actor_id),
                 timestamp: std::time::SystemTime::now(),
@@ -3207,7 +3207,7 @@ impl<A: Actor> Instance<A> {
             let from_actor_id = headers
                 .get(crate::mailbox::headers::SENDER_ACTOR_ID_HASH)
                 .unwrap_or(0);
-            let to_actor_id = hash_to_u64(self.self_addr());
+            let to_actor_id = hash_to_u64(self.self_addr().id());
             let port_id = headers.get(crate::mailbox::headers::TELEMETRY_PORT_ID);
 
             notify_message(hyperactor_telemetry::MessageEvent {

hyperactor_mesh/src/actor_mesh.rs

@@ -537,7 +537,7 @@ impl<A: Referable> ActorMeshRef<A> {
 
         hyperactor_telemetry::notify_sent_message(hyperactor_telemetry::SentMessageEvent {
             timestamp: std::time::SystemTime::now(),
-            sender_actor_id: hyperactor_telemetry::hash_to_u64(cx.mailbox().actor_addr()),
+            sender_actor_id: hyperactor_telemetry::hash_to_u64(cx.mailbox().actor_addr().id()),
             actor_mesh_id: hyperactor_telemetry::hash_to_u64(&self.id.to_string()),
             view_json: serde_json::to_string(view::Ranked::region(self)).unwrap_or_default(),
             shape_json: {

hyperactor_mesh/src/comm/multicast.rs

@@ -371,7 +371,7 @@ pub fn set_cast_info_on_headers(headers: &mut Flattrs, cast_point: Point, sender
     // CAST_ORIGINATING_SENDER -- they carry overlapping sender identity.
     headers.set(
         hyperactor::mailbox::headers::SENDER_ACTOR_ID_HASH,
-        hyperactor_telemetry::hash_to_u64(&sender),
+        hyperactor_telemetry::hash_to_u64(sender.id()),
     );
     headers.set(CAST_POINT, cast_point);
     headers.set(CAST_ORIGINATING_SENDER, sender);

hyperactor_mesh/src/host_mesh.rs

@@ -405,7 +405,7 @@ impl HostMesh {
         for (rank, host) in self.current_ref.hosts().iter().enumerate() {
             let actor = host.mesh_agent();
             hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
-                id: hyperactor_telemetry::hash_to_u64(&actor.actor_addr()),
+                id: hyperactor_telemetry::hash_to_u64(actor.actor_addr().id()),
                 timestamp: now,
                 mesh_id: mesh_id_hash,
                 rank: rank as u64,

hyperactor_mesh/src/proc_mesh.rs

@@ -205,14 +205,14 @@ impl ProcMesh {
             // These are skipped in Proc::spawn_inner. mesh_id directly points to proc mesh.
             let now = std::time::SystemTime::now();
             for rank in current_ref.ranks.iter() {
-                let actor_id = rank.agent.actor_addr();
+                let actor_addr = rank.agent.actor_addr();
 
                 hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
-                    id: hyperactor_telemetry::hash_to_u64(&actor_id),
+                    id: hyperactor_telemetry::hash_to_u64(actor_addr.id()),
                     timestamp: now,
                     mesh_id: mesh_id_hash,
                     rank: rank.create_rank as u64,
-                    full_name: actor_id.to_string(),
+                    full_name: actor_addr.to_string(),
                     display_name: None,
                 });
             }
@@ -883,13 +883,13 @@ impl ProcMeshRef {
                     let point = self.region().extent().point_of_rank(rank).unwrap();
                     crate::actor_display_name(sdn, &point)
                 });
-                let actor_id = proc_ref.actor_addr(&actor_mesh_id);
+                let actor_addr = proc_ref.actor_addr(&actor_mesh_id);
                 hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
-                    id: hyperactor_telemetry::hash_to_u64(&actor_id),
+                    id: hyperactor_telemetry::hash_to_u64(actor_addr.id()),
                     timestamp: now,
                     mesh_id: mesh_id_hash,
                     rank: rank as u64,
-                    full_name: actor_id.to_string(),
+                    full_name: actor_addr.to_string(),
                     display_name,
                 });
             }

hyperactor_telemetry/src/lib.rs

@@ -111,6 +111,7 @@ use crate::config::ENABLE_RECORDER_TRACING;
 use crate::config::ENABLE_SQLITE_TRACING;
 use crate::config::MONARCH_LOG_SUFFIX;
 use crate::recorder::Recorder;
+
 /// Hash any hashable value to a u64 using DefaultHasher.
 pub fn hash_to_u64(value: &impl Hash) -> u64 {
     let mut hasher = DefaultHasher::new();
@@ -335,11 +336,11 @@ pub fn end_user_span(id: u64) {
 /// Event data for actor creation.
 #[derive(Debug, Clone)]
 pub struct ActorEvent {
-    /// Unique identifier for this actor (hashed from ActorAddr)
+    /// Unique identifier for this actor, hashed from ActorId.
     pub id: u64,
     /// Timestamp when the actor was created
     pub timestamp: SystemTime,
-    /// ID of the mesh this actor belongs to (hashed from actor_name)
+    /// ID of the mesh this actor belongs to, matching `MeshEvent.id`.
     pub mesh_id: u64,
     /// Rank index into the mesh shape
     pub rank: u64,
@@ -407,7 +408,7 @@ pub fn notify_actor_status_changed(event: ActorStatusEvent) {
 #[derive(Debug, Clone)]
 pub struct SentMessageEvent {
     pub timestamp: SystemTime,
-    /// Hash of the sending actor's [`ActorAddr`].
+    /// Hash of the sending actor's ActorId.
     pub sender_actor_id: u64,
     /// Hash of the target actor mesh's name.
     pub actor_mesh_id: u64,
@@ -432,9 +433,9 @@ pub struct MessageEvent {
     pub timestamp: SystemTime,
     /// Unique identifier for this received message.
     pub id: u64,
-    /// Hash of sender's ActorAddr.
+    /// Hash of sender's ActorId.
     pub from_actor_id: u64,
-    /// Hash of receiver's ActorAddr.
+    /// Hash of receiver's ActorId.
     pub to_actor_id: u64,
     /// Endpoint name if this message targets a specific actor endpoint
     pub endpoint: Option<String>,

monarch_hyperactor/src/host_mesh.rs

@@ -425,13 +425,13 @@ fn bootstrap_host(bootstrap_cmd: Option<PyBootstrapCommand>) -> PyResult<PyPytho
                 parent_view_json: None,
             });
 
-            let host_agent_id = host_mesh_agent.actor_addr();
+            let host_agent_addr = host_mesh_agent.actor_addr();
             hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
-                id: hyperactor_telemetry::hash_to_u64(host_agent_id),
+                id: hyperactor_telemetry::hash_to_u64(host_agent_addr.id()),
                 timestamp: now,
                 mesh_id: host_mesh_id,
                 rank: 0,
-                full_name: host_agent_id.to_string(),
+                full_name: host_agent_addr.to_string(),
                 display_name: None,
             });
 
@@ -453,13 +453,13 @@ fn bootstrap_host(bootstrap_cmd: Option<PyBootstrapCommand>) -> PyResult<PyPytho
                 parent_view_json: None,
             });
 
-            let proc_agent_id = local_proc_agent.actor_addr();
+            let proc_agent_addr = local_proc_agent.actor_addr();
             hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
-                id: hyperactor_telemetry::hash_to_u64(proc_agent_id),
+                id: hyperactor_telemetry::hash_to_u64(proc_agent_addr.id()),
                 timestamp: now,
                 mesh_id: proc_mesh_id,
                 rank: 0,
-                full_name: proc_agent_id.to_string(),
+                full_name: proc_agent_addr.to_string(),
                 display_name: None,
             });
 
@@ -477,7 +477,7 @@ fn bootstrap_host(bootstrap_cmd: Option<PyBootstrapCommand>) -> PyResult<PyPytho
             });
 
             hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
-                id: hyperactor_telemetry::hash_to_u64(instance.self_addr()),
+                id: hyperactor_telemetry::hash_to_u64(instance.self_addr().id()),
                 timestamp: now,
                 mesh_id: client_mesh_id,
                 rank: 0,

python/tests/test_distributed_telemetry.py

@@ -668,6 +668,17 @@ def test_actor_status_events_table() -> None:
             f"Expected at least one actor status event, got {event_count}"
         )
 
+        joined = engine.query(
+            "SELECT ase.id FROM actor_status_events ase "
+            "INNER JOIN actors a ON ase.actor_id = a.id "
+            "WHERE a.display_name LIKE '%status_test_worker%'"
+        )
+        joined_count = len(joined.to_pydict().get("id", []))
+        assert joined_count > 0, (
+            "Expected actor_status_events.actor_id to join actors.id for "
+            f"status_test_worker, got {joined_count} joined rows"
+        )
+
         # Verify new_status values are valid ActorStatus arm names
         valid_statuses = {
             "Unknown",
@@ -795,7 +806,7 @@ def test_sent_messages_table(
     All send paths (call, call_one, broadcast, choose) go through
     cast_with_selection in actor_mesh.rs, which calls notify_sent_message
     with a SentMessageEvent containing:
-      - sender_actor_id: hash of the sending actor's ActorAddr
+      - sender_actor_id: hash of the sending actor's ActorId
       - actor_mesh_id: hash of the target actor mesh name
       - view_json: serialized ndslice::Region of the current view
       - shape_json: serialized ndslice::Shape (converted from the Region)