fix(runtime): give event publishers unique discovery identities

lib/runtime/src/discovery/kube.rs

@@ -129,11 +129,11 @@ impl Discovery for KubeDiscoveryClient {
     }
 
     async fn register_internal(&self, spec: DiscoverySpec) -> Result<DiscoveryInstance> {
-        let instance_id = self.instance_id();
-        let instance = spec.with_instance_id(instance_id);
+        let instance = spec.into_instance(self.instance_id());
+        let instance_id = instance.instance_id();
 
         tracing::debug!(
-            "Registering instance: {:?} with instance_id={:x}",
+            "Registering discovery instance: {:?}, instance_id={:x}",
             instance,
             instance_id
         );
@@ -214,7 +214,7 @@ impl Discovery for KubeDiscoveryClient {
     }
 
     async fn unregister(&self, instance: DiscoveryInstance) -> Result<()> {
-        let instance_id = self.instance_id();
+        let instance_id = instance.instance_id();
 
         // Write to local metadata and persist to CR
         // IMPORTANT: Hold the write lock across the CR write to prevent race conditions

lib/runtime/src/discovery/kv_store.rs

@@ -416,8 +416,8 @@ impl Discovery for KVStoreDiscovery {
     }
 
     async fn register_internal(&self, spec: DiscoverySpec) -> Result<DiscoveryInstance> {
-        let instance_id = self.instance_id();
-        let instance = spec.with_instance_id(instance_id);
+        let instance = spec.into_instance(self.instance_id());
+        let instance_id = instance.instance_id();
 
         let (bucket_name, key_path) = match &instance {
             DiscoveryInstance::Endpoint(inst) => {
@@ -528,7 +528,7 @@ impl Discovery for KVStoreDiscovery {
         // Use revision 0 for initial registration
         let outcome = bucket.insert(&key, instance_json.into(), 0).await?;
         tracing::debug!(
-            "KVStoreDiscovery::register: Successfully registered instance_id={}, key={}, outcome={:?}",
+            "KVStoreDiscovery::register: Registration insert completed instance_id={}, key={}, outcome={:?}",
             instance_id,
             key_path,
             outcome

lib/runtime/src/discovery/mock.rs

@@ -158,7 +158,7 @@ impl Discovery for MockDiscovery {
     }
 
     async fn register_internal(&self, spec: DiscoverySpec) -> Result<DiscoveryInstance> {
-        let instance = spec.with_instance_id(self.instance_id);
+        let instance = spec.into_instance(self.instance_id);
 
         self.registry
             .instances

lib/runtime/src/discovery/mod.rs

@@ -350,6 +350,11 @@ pub enum DiscoverySpec {
         component: String,
         /// Topic name for this channel (e.g., "kv-events", "kv-metrics")
         topic: String,
+        /// Unique identity of this publisher incarnation.
+        ///
+        /// A process can host multiple publishers for the same topic, so event
+        /// channels cannot use the process-level discovery instance ID.
+        publisher_id: u64,
         /// Event transport type (NATS subject prefix or ZMQ endpoint)
         transport: EventTransport,
     },
@@ -392,8 +397,12 @@ impl DiscoverySpec {
         })
     }
 
-    /// Attaches an instance ID to create a DiscoveryInstance
-    pub fn with_instance_id(self, instance_id: u64) -> DiscoveryInstance {
+    /// Converts this registration spec into a discovery instance.
+    ///
+    /// Endpoint and model specs use `default_instance_id`, normally the
+    /// discovery client's process-level ID. Event channel specs already carry
+    /// a publisher-level ID, so they use that instead.
+    pub fn into_instance(self, default_instance_id: u64) -> DiscoveryInstance {
         match self {
             Self::Endpoint {
                 namespace,
@@ -405,7 +414,7 @@ impl DiscoverySpec {
                 namespace,
                 component,
                 endpoint,
-                instance_id,
+                instance_id: default_instance_id,
                 transport,
                 device_type,
             }),
@@ -419,24 +428,30 @@ impl DiscoverySpec {
                 namespace,
                 component,
                 endpoint,
-                instance_id,
+                instance_id: default_instance_id,
                 card_json,
                 model_suffix,
             },
             Self::EventChannel {
                 namespace,
                 component,
                 topic,
+                publisher_id,
                 transport,
             } => DiscoveryInstance::EventChannel {
                 namespace,
                 component,
                 topic,
-                instance_id,
+                instance_id: publisher_id,
                 transport,
             },
         }
     }
+
+    /// Compatibility alias for [`DiscoverySpec::into_instance`].
+    pub fn with_instance_id(self, default_instance_id: u64) -> DiscoveryInstance {
+        self.into_instance(default_instance_id)
+    }
 }
 
 /// Registered instances in the discovery plane
@@ -789,7 +804,9 @@ fn find_conflicting_model_name(
 #[async_trait]
 pub trait Discovery: Send + Sync {
     /// Returns a unique identifier for this worker (e.g lease id if using etcd or generated id for memory store)
-    /// Discovery objects created by this worker will be associated with this id.
+    /// Endpoint and model objects created by this worker use this ID. Event
+    /// channels use a publisher-level ID because a worker can own more than one
+    /// publisher for the same topic.
     fn instance_id(&self) -> u64;
 
     /// Registers an object in the discovery plane with the instance id

lib/runtime/src/transports/event_plane/dynamic_subscriber.rs

@@ -58,8 +58,9 @@ impl DynamicSubscriber {
         let (event_tx, event_rx) = mpsc::channel::<Bytes>(channel_cap);
 
         // Track active endpoint connections with instance ID to endpoint mapping
-        let active_endpoints: Arc<RwLock<HashMap<String, (String, CancellationToken)>>> =
-            Arc::new(RwLock::new(HashMap::new()));
+        let active_endpoints: Arc<
+            RwLock<HashMap<DiscoveryInstanceId, (String, CancellationToken)>>,
+        > = Arc::new(RwLock::new(HashMap::new()));
 
         // Clone self for the spawned task
         let subscriber_clone = Arc::clone(&self);
@@ -103,19 +104,19 @@ impl DynamicSubscriber {
                 match event_result {
                     Ok(DiscoveryEvent::Added(instance)) => {
                         tracing::info!(instance = ?instance, "Discovery Added event received");
-                        let instance_id = instance.instance_id().to_string();
+                        let instance_id = instance.id();
 
                         // Extract ZMQ endpoint from the instance
-                        if let Some(endpoint) = Self::extract_zmq_endpoint(&instance) {
+                        if let Some(endpoint) = Self::extract_zmq_endpoint(&instance, &zmq_topic) {
                             let mut endpoints_guard = endpoints.write().await;
 
                             // Skip if instance already tracked
                             if endpoints_guard.contains_key(&instance_id) {
-                                tracing::debug!(endpoint = %endpoint, instance_id = %instance_id, "Already connected to ZMQ publisher");
+                                tracing::debug!(endpoint = %endpoint, ?instance_id, "Already connected to ZMQ publisher");
                                 continue;
                             }
 
-                            tracing::info!(endpoint = %endpoint, instance_id = %instance_id, "Connecting to new ZMQ publisher");
+                            tracing::info!(endpoint = %endpoint, ?instance_id, "Connecting to new ZMQ publisher");
 
                             // Create cancellation token for this endpoint's stream
                             let endpoint_cancel = CancellationToken::new();
@@ -151,25 +152,44 @@ impl DynamicSubscriber {
                                 endpoints_clone.write().await.remove(&instance_id_clone);
                             });
                         } else {
-                            tracing::warn!(
+                            tracing::debug!(
                                 instance = ?instance,
-                                "Discovery Added event did not contain a ZMQ endpoint"
+                                expected_topic = %zmq_topic,
+                                "Discovery event is not a matching ZMQ publisher"
                             );
                         }
                     }
                     Ok(DiscoveryEvent::Removed(instance_id)) => {
-                        let id_str = instance_id.instance_id().to_string();
+                        let is_expected_topic = matches!(
+                            &instance_id,
+                            DiscoveryInstanceId::EventChannel(channel_id)
+                                if channel_id.topic == zmq_topic
+                        );
+                        if !is_expected_topic {
+                            tracing::debug!(
+                                ?instance_id,
+                                expected_topic = %zmq_topic,
+                                "Ignoring removal for unrelated event channel"
+                            );
+                            continue;
+                        }
+
                         tracing::info!(
-                            instance_id = %id_str,
+                            ?instance_id,
                             "ZMQ publisher removed from discovery, cancelling endpoint stream"
                         );
 
                         // Cancel the endpoint's stream via its CancellationToken
-                        if let Some((_endpoint, cancel)) = endpoints.write().await.remove(&id_str) {
+                        if let Some((_endpoint, cancel)) =
+                            endpoints.write().await.remove(&instance_id)
+                        {
                             cancel.cancel();
-                            tracing::info!(instance_id = %id_str, "Cancelled endpoint stream");
+                            tracing::info!(?instance_id, "Cancelled endpoint stream");
                         } else {
-                            tracing::warn!(instance_id = %id_str, "No active endpoint found for removed stream instance");
+                            tracing::debug!(
+                                ?instance_id,
+                                "No active endpoint found for removed stream instance"
+                            );
                         }
                     }
                     Err(e) => {
@@ -201,8 +221,11 @@ impl DynamicSubscriber {
     }
 
     /// Extract ZMQ endpoint from a discovery instance.
-    fn extract_zmq_endpoint(instance: &DiscoveryInstance) -> Option<String> {
-        if let DiscoveryInstance::EventChannel { transport, .. } = instance
+    fn extract_zmq_endpoint(instance: &DiscoveryInstance, expected_topic: &str) -> Option<String> {
+        if let DiscoveryInstance::EventChannel {
+            topic, transport, ..
+        } = instance
+            && topic == expected_topic
             && let EventTransport::Zmq { endpoint } = transport
         {
             return Some(endpoint.clone());
@@ -278,3 +301,41 @@ impl Drop for DynamicSubscriber {
         self.cancel_token.cancel();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn event_channel(topic: &str, transport: EventTransport) -> DiscoveryInstance {
+        DiscoveryInstance::EventChannel {
+            namespace: "test-ns".to_string(),
+            component: "test-component".to_string(),
+            topic: topic.to_string(),
+            instance_id: 1,
+            transport,
+        }
+    }
+
+    #[test]
+    fn extracts_only_matching_zmq_topic() {
+        let matching = event_channel("kv-events", EventTransport::zmq("tcp://127.0.0.1:1"));
+        let wrong_topic = event_channel("kv-metrics", EventTransport::zmq("tcp://127.0.0.1:2"));
+        let wrong_transport = event_channel(
+            "kv-events",
+            EventTransport::nats("namespace.test-ns.component.test-component"),
+        );
+
+        assert_eq!(
+            DynamicSubscriber::extract_zmq_endpoint(&matching, "kv-events").as_deref(),
+            Some("tcp://127.0.0.1:1")
+        );
+        assert_eq!(
+            DynamicSubscriber::extract_zmq_endpoint(&wrong_topic, "kv-events"),
+            None
+        );
+        assert_eq!(
+            DynamicSubscriber::extract_zmq_endpoint(&wrong_transport, "kv-events"),
+            None
+        );
+    }
+}