fix(runtime): preserve event channel metadata compatibility [DYN-3846]

lib/runtime/src/discovery/kube/crd.rs

@@ -17,7 +17,7 @@ use kube::{
 };
 use serde::{Deserialize, Serialize};
 
-use crate::discovery::DiscoveryMetadata;
+use crate::discovery::{DiscoveryMetadata, EventScope};
 
 /// Field manager name for server-side apply - identifies this client as the owner of fields it sets
 const FIELD_MANAGER: &str = "dynamo-worker";
@@ -58,7 +58,8 @@ pub fn build_cr(
     pod_uid: &str,
     metadata: &DiscoveryMetadata,
 ) -> Result<DynamoWorkerMetadata> {
-    let data = serde_json::to_value(metadata)?;
+    let mut data = serde_json::to_value(metadata)?;
+    add_legacy_event_channel_fields(&mut data)?;
     let spec = DynamoWorkerMetadataSpec::new(data);
     let mut cr = DynamoWorkerMetadata::new(cr_name, spec);
 
@@ -78,6 +79,81 @@ pub fn build_cr(
     Ok(cr)
 }
 
+/// Accept the pre-scope event-channel shape at the Kubernetes DWM boundary.
+pub(super) fn deserialize_metadata(mut data: serde_json::Value) -> Result<DiscoveryMetadata> {
+    add_current_event_channel_scopes(&mut data)?;
+    Ok(serde_json::from_value(data)?)
+}
+
+fn add_legacy_event_channel_fields(data: &mut serde_json::Value) -> Result<()> {
+    let Some(channels) = data
+        .get_mut("event_channels")
+        .and_then(serde_json::Value::as_object_mut)
+    else {
+        return Ok(());
+    };
+
+    for channel in channels.values_mut() {
+        let Some(channel) = channel.as_object_mut() else {
+            continue;
+        };
+        let Some(scope) = channel.get("scope").cloned() else {
+            continue;
+        };
+        let scope = serde_json::from_value::<EventScope>(scope)?;
+
+        channel.insert(
+            "namespace".to_string(),
+            serde_json::Value::String(scope.namespace().to_string()),
+        );
+        channel.insert(
+            "component".to_string(),
+            // v1.2 represented namespace-scoped publishers with an empty component.
+            serde_json::Value::String(scope.component().unwrap_or("").to_string()),
+        );
+    }
+
+    Ok(())
+}
+
+fn add_current_event_channel_scopes(data: &mut serde_json::Value) -> Result<()> {
+    let Some(channels) = data
+        .get_mut("event_channels")
+        .and_then(serde_json::Value::as_object_mut)
+    else {
+        return Ok(());
+    };
+
+    for channel in channels.values_mut() {
+        let Some(channel) = channel.as_object_mut() else {
+            continue;
+        };
+        if channel.contains_key("scope") {
+            continue;
+        }
+        let (Some(namespace), Some(component)) = (
+            channel.get("namespace").and_then(serde_json::Value::as_str),
+            channel.get("component").and_then(serde_json::Value::as_str),
+        ) else {
+            continue;
+        };
+        let scope = if component.is_empty() {
+            EventScope::Namespace {
+                name: namespace.to_string(),
+            }
+        } else {
+            EventScope::Component {
+                namespace: namespace.to_string(),
+                component: component.to_string(),
+            }
+        };
+
+        channel.insert("scope".to_string(), serde_json::to_value(scope)?);
+    }
+
+    Ok(())
+}
+
 /// Apply (create or update) a DynamoWorkerMetadata CR using server-side apply
 ///
 /// This function uses Kubernetes server-side apply which:
@@ -123,7 +199,8 @@ pub async fn apply_cr(
 mod tests {
     use super::*;
     use crate::discovery::{
-        DiscoveryInstance, DiscoveryQuery, EventScope, EventSourceQuery, MAX_JSON_SAFE_PUBLISHER_ID,
+        DiscoveryInstance, DiscoveryQuery, EventChannelQuery, EventScope, EventSourceQuery,
+        EventTransport, MAX_JSON_SAFE_PUBLISHER_ID,
     };
     use crate::protocols::EndpointId;
     use kube::Resource;
@@ -191,7 +268,7 @@ mod tests {
             .expect("serialized event source publisher ID");
         assert_eq!(publisher_id.as_u64(), Some(MAX_JSON_SAFE_PUBLISHER_ID));
 
-        let round_trip: DiscoveryMetadata = serde_json::from_value(cr.spec.data).unwrap();
+        let round_trip = deserialize_metadata(cr.spec.data).unwrap();
 
         assert_eq!(
             round_trip.filter(&DiscoveryQuery::EventSources(
@@ -200,4 +277,121 @@ mod tests {
             vec![source]
         );
     }
+
+    #[test]
+    fn event_channel_metadata_supports_v1_2_wire_shape() {
+        #[derive(serde::Deserialize)]
+        #[serde(tag = "type")]
+        enum LegacyDiscoveryInstance {
+            EventChannel {
+                namespace: String,
+                component: String,
+                topic: String,
+                instance_id: u64,
+                transport: EventTransport,
+            },
+        }
+
+        let endpoint = EndpointId {
+            namespace: "workers".to_string(),
+            component: "backend".to_string(),
+            name: "generate".to_string(),
+        };
+        let transport = EventTransport::zmq("tcp://worker:5555");
+        let channel = DiscoveryInstance::EventChannel {
+            scope: EventScope::Endpoint {
+                endpoint: endpoint.clone(),
+            },
+            topic: "kv-events".to_string(),
+            instance_id: 42,
+            transport: transport.clone(),
+        };
+        let mut metadata = DiscoveryMetadata::new();
+        metadata.register_event_channel(channel.clone()).unwrap();
+
+        let cr = build_cr("test-pod", "test-pod", "pod-uid", &metadata).unwrap();
+        let encoded_channel = cr.spec.data["event_channels"]
+            .as_object()
+            .and_then(|channels| channels.values().next())
+            .cloned()
+            .expect("serialized event channel");
+        let LegacyDiscoveryInstance::EventChannel {
+            namespace,
+            component,
+            topic,
+            instance_id,
+            transport: legacy_transport,
+        } = serde_json::from_value(encoded_channel).expect("v1.2 event channel shape");
+        assert_eq!(
+            (namespace, component, topic, instance_id, legacy_transport),
+            (
+                "workers".to_string(),
+                "backend".to_string(),
+                "kv-events".to_string(),
+                42,
+                transport.clone(),
+            )
+        );
+        let current_round_trip = deserialize_metadata(cr.spec.data).unwrap();
+        assert_eq!(
+            current_round_trip.filter(&DiscoveryQuery::EventChannels(
+                EventChannelQuery::endpoint_topic(endpoint, "kv-events")
+            )),
+            vec![channel]
+        );
+
+        let namespace_transport = EventTransport::nats("namespace.workers");
+        let legacy_metadata = serde_json::json!({
+            "endpoints": {},
+            "model_cards": {},
+            "event_channels": {
+                "workers/backend/kv-events/2a": {
+                    "type": "EventChannel",
+                    "namespace": "workers",
+                    "component": "backend",
+                    "topic": "kv-events",
+                    "instance_id": 42,
+                    "transport": transport,
+                },
+                "workers//namespace-events/2b": {
+                    "type": "EventChannel",
+                    "namespace": "workers",
+                    "component": "",
+                    "topic": "namespace-events",
+                    "instance_id": 43,
+                    "transport": namespace_transport,
+                }
+            }
+        });
+        let upgraded = deserialize_metadata(legacy_metadata).unwrap();
+        assert_eq!(
+            upgraded.filter(&DiscoveryQuery::EventChannels(EventChannelQuery::topic(
+                "workers",
+                "backend",
+                "kv-events",
+            ))),
+            vec![DiscoveryInstance::EventChannel {
+                scope: EventScope::Component {
+                    namespace: "workers".to_string(),
+                    component: "backend".to_string(),
+                },
+                topic: "kv-events".to_string(),
+                instance_id: 42,
+                transport,
+            }]
+        );
+        assert_eq!(
+            upgraded.filter(&DiscoveryQuery::EventChannels(
+                EventChannelQuery::namespace_topic("workers", "namespace-events")
+            )),
+            vec![DiscoveryInstance::EventChannel {
+                scope: EventScope::Namespace {
+                    name: "workers".to_string(),
+                },
+                topic: "namespace-events".to_string(),
+                instance_id: 43,
+                transport: namespace_transport,
+            }]
+        );
+    }
 }

lib/runtime/src/discovery/kube/daemon.rs

@@ -324,7 +324,7 @@ impl DiscoveryDaemon {
                 continue;
             }
 
-            match serde_json::from_value::<DiscoveryMetadata>(arc_cr.spec.data.clone()) {
+            match super::crd::deserialize_metadata(arc_cr.spec.data.clone()) {
                 Ok(metadata) => {
                     tracing::trace!("Loaded metadata from CR '{cr_name}'");
                     let cached = CachedCrMetadata {