Revisit mesh telemetry IDs

Summary:
ActorMeshId values are only unique within a ProcMesh, so actor mesh telemetry IDs must include both `ProcMeshId` and `ActorMeshId`. This showed up with the sidecar design because entity memtables are now per host: singleton actor mesh IDs from multiple proc meshes on the same host land in the same table, so hashing only the actor mesh name makes those singleton rows collide. Previously, this inconsistency was hidden. Add `telemetry_actor_mesh_id` as the shared derivation helper, and use it for actor mesh creation, sent-message telemetry, actor rows, and the bootstrap client mesh so `sent_messages.actor_mesh_id`, `meshes.id`, and `actors.mesh_id` stay joinable. Also align host and proc mesh row IDs to hash typed `HostMeshId` and `ProcMeshId` values, so `meshes.parent_mesh_id` uses the same derivation as the parent `meshes.id` row.

Caveat: this does not fully address potential collisions for `ProcMeshId::singleton` values themselves; proc mesh telemetry IDs may also need parent scoping if multiple singleton proc meshes can share a host-local memtable.

Differential Revision: D107178870

Author: thedavekwon

hyperactor_mesh/src/actor_mesh.rs

@@ -75,6 +75,7 @@ use crate::mesh_id::ActorMeshId;
 use crate::metrics;
 use crate::proc_agent::ActorState;
 use crate::proc_mesh::GET_ACTOR_STATE_MAX_IDLE;
+use crate::proc_mesh::telemetry_actor_mesh_id;
 use crate::resource;
 use crate::supervision::MeshFailure;
 use crate::supervision::Unhealthy;
@@ -538,7 +539,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().id()),
-            actor_mesh_id: hyperactor_telemetry::hash_to_u64(&self.id.to_string()),
+            actor_mesh_id: telemetry_actor_mesh_id(self.proc_mesh.id(), &self.id),
             view_json: serde_json::to_string(view::Ranked::region(self)).unwrap_or_default(),
             shape_json: {
                 let shape: ndslice::Shape = view::Ranked::region(self).into();

hyperactor_mesh/src/host_mesh.rs

@@ -381,7 +381,7 @@ impl HostMesh {
     /// Emit a telemetry event for this host mesh creation.
     fn notify_created(&self) {
         let name_str = self.id.to_string();
-        let mesh_id_hash = hyperactor_telemetry::hash_to_u64(&name_str);
+        let mesh_id_hash = hyperactor_telemetry::hash_to_u64(&self.id);
 
         hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
             id: mesh_id_hash,

hyperactor_mesh/src/proc_mesh.rs

@@ -32,6 +32,7 @@ use hyperactor::supervision::ActorSupervisionEvent;
 use hyperactor_config::CONFIG;
 use hyperactor_config::ConfigAttr;
 use hyperactor_config::attrs::declare_attrs;
+use hyperactor_telemetry::hash_to_u64;
 use ndslice::Extent;
 use ndslice::ViewExt as _;
 use ndslice::view;
@@ -86,6 +87,11 @@ declare_attrs! {
 /// present as a system actor (`system_children`) on every proc mesh member.
 pub const COMM_ACTOR_NAME: &str = "comm";
 
+/// Returns the telemetry `meshes.id` value for an actor mesh.
+pub fn telemetry_actor_mesh_id(proc_mesh_id: &ProcMeshId, actor_mesh_id: &ActorMeshId) -> u64 {
+    hash_to_u64(&(proc_mesh_id, actor_mesh_id))
+}
+
 /// A reference to a single [`hyperactor::Proc`].
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct ProcRef {
@@ -176,13 +182,13 @@ impl ProcMesh {
         // Notify telemetry that the ProcAgent mesh was created.
         {
             let name_str = id.to_string();
-            let mesh_id_hash = hyperactor_telemetry::hash_to_u64(&name_str);
+            let mesh_id_hash = hash_to_u64(&id);
 
             let hm = current_ref
                 .host_mesh
                 .as_ref()
                 .expect("ProcMesh always has a host mesh");
-            let parent_mesh_id = hyperactor_telemetry::hash_to_u64(&hm.id().to_string());
+            let parent_mesh_id = hash_to_u64(hm.id());
             let parent_view_json = serde_json::to_string(hm.region())
                 .unwrap_or_else(|e| format!("encountered error when serializing region: {}", e));
 
@@ -848,12 +854,9 @@ impl ProcMeshRef {
         {
             let id_str = mesh.id().to_string();
 
-            // Hash the actor mesh id. This is used as mesh_id for both
-            // the MeshEvent and the per-actor ActorEvents below.
-            let mesh_id_hash = hyperactor_telemetry::hash_to_u64(&id_str);
-
             // Hash the proc mesh id for parent_mesh_id.
-            let parent_mesh_id_hash = hyperactor_telemetry::hash_to_u64(&self.id().to_string());
+            let parent_mesh_id_hash = hash_to_u64(self.id());
+            let mesh_id_hash = telemetry_actor_mesh_id(self.id(), mesh.id());
 
             hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
                 id: mesh_id_hash,

hyperactor_telemetry/src/lib.rs

@@ -410,7 +410,7 @@ pub struct SentMessageEvent {
     pub timestamp: SystemTime,
     /// Hash of the sending actor's ActorId.
     pub sender_actor_id: u64,
-    /// Hash of the target actor mesh's name.
+    /// Hash of the target actor mesh's `(ProcMeshId, ActorMeshId)`.
     pub actor_mesh_id: u64,
     /// The view (slice) of the actor mesh that was targeted, serialized from
     /// [`ndslice::Region`]. For full-mesh sends (call, broadcast) this covers

monarch_hyperactor/src/host_mesh.rs

@@ -29,12 +29,15 @@ use hyperactor_mesh::host_mesh::host_agent::GetLocalProcClient;
 use hyperactor_mesh::host_mesh::host_agent::HostAgent;
 use hyperactor_mesh::host_mesh::host_agent::ShutdownHost;
 use hyperactor_mesh::mesh_admin::MeshAdminMessageClient;
+use hyperactor_mesh::mesh_id::ActorMeshId;
 use hyperactor_mesh::mesh_id::HostMeshId;
 use hyperactor_mesh::mesh_id::ProcMeshId;
 use hyperactor_mesh::proc_agent::GetProcClient;
 use hyperactor_mesh::proc_mesh::ProcRef;
+use hyperactor_mesh::proc_mesh::telemetry_actor_mesh_id;
 use hyperactor_mesh::shared_cell::SharedCell;
 use hyperactor_mesh::transport::default_bind_spec;
+use hyperactor_telemetry::hash_to_u64;
 use ndslice::View;
 use ndslice::view::RankedSliceable;
 use pyo3::IntoPyObjectExt;
@@ -408,7 +411,7 @@ fn bootstrap_host(bootstrap_cmd: Option<PyBootstrapCommand>) -> PyResult<PyPytho
             let now = std::time::SystemTime::now();
 
             let host_name_str = host_mesh.id().to_string();
-            let host_mesh_id = hyperactor_telemetry::hash_to_u64(&host_name_str);
+            let host_mesh_id = hash_to_u64(host_mesh.id());
             hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
                 id: host_mesh_id,
                 timestamp: now,
@@ -436,7 +439,7 @@ fn bootstrap_host(bootstrap_cmd: Option<PyBootstrapCommand>) -> PyResult<PyPytho
             });
 
             let proc_id_str = proc_mesh.id().to_string();
-            let proc_mesh_id = hyperactor_telemetry::hash_to_u64(&proc_id_str);
+            let proc_mesh_id = hash_to_u64(proc_mesh.id());
             hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
                 id: proc_mesh_id,
                 timestamp: now,
@@ -463,8 +466,9 @@ fn bootstrap_host(bootstrap_cmd: Option<PyBootstrapCommand>) -> PyResult<PyPytho
                 display_name: None,
             });
 
+            let client_mesh_actor_id = ActorMeshId::singleton(Label::new("client").unwrap());
             let client_mesh_name = format!("{}/client", proc_mesh.id());
-            let client_mesh_id = hyperactor_telemetry::hash_to_u64(&client_mesh_name);
+            let client_mesh_id = telemetry_actor_mesh_id(proc_mesh.id(), &client_mesh_actor_id);
             hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
                 id: client_mesh_id,
                 timestamp: now,

python/tests/test_distributed_telemetry.py

@@ -807,7 +807,7 @@ def test_sent_messages_table(
     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 ActorId
-      - actor_mesh_id: hash of the target actor mesh name
+      - actor_mesh_id: hash of the target (ProcMeshId, ActorMeshId)
       - view_json: serialized ndslice::Region of the current view
       - shape_json: serialized ndslice::Shape (converted from the Region)
     """
@@ -860,6 +860,25 @@ def test_sent_messages_table(
             f"Expected 42 sent_messages via {send_path}, got {joined_count}"
         )
 
+        actor_joined = engine.query(
+            "SELECT COUNT(DISTINCT sm.id) AS message_count, "
+            "COUNT(DISTINCT a.id) AS actor_count "
+            "FROM sent_messages sm "
+            "JOIN actors a ON sm.actor_mesh_id = a.mesh_id "
+            "JOIN meshes m ON a.mesh_id = m.id "
+            f"WHERE m.given_name = '{mesh_name}'"
+        )
+        actor_joined_dict = actor_joined.to_pydict()
+        joined_message_count = actor_joined_dict["message_count"][0]
+        joined_actor_count = actor_joined_dict["actor_count"][0]
+        assert joined_message_count == 42, (
+            "Expected sent_messages.actor_mesh_id to join actors.mesh_id for "
+            f"{send_path}, got {joined_message_count} messages"
+        )
+        assert joined_actor_count == 2, (
+            f"Expected 2 target actors for {send_path}, got {joined_actor_count}"
+        )
+
         # Verify view_json (ndslice Region) and shape_json (ndslice Shape).
         # Region serializes as {"labels": [...], "slice": {"offset": ..., "sizes": [...], "strides": [...]}}.
         # Shape is Region converted via Region::into::<Shape>, same serialization format.