Skip to content

fix(runtime): preserve event channel metadata compatibility (cherry-pick #12708) #12840

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pvijayakrish merged 1 commit into from
Aug 7, 2026
Merged

fix(runtime): preserve event channel metadata compatibility (cherry-pick #12708) #12840

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
202 changes: 198 additions & 4 deletions lib/runtime/src/discovery/kube/crd.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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);

Expand All @@ -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)?)
}
Comment on lines +82 to +86

@devin-ai-integration devin-ai-integration Bot Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 Compatibility shim applies only to the Kubernetes DWM path, not the KV-store discovery backend

The legacy field emission/reconstruction lives entirely in build_cr/deserialize_metadata for the Kubernetes CR boundary. lib/runtime/src/discovery/kv_store.rs serializes DiscoveryInstance directly (serde_json::to_vec(&instance) at lib/runtime/src/discovery/kv_store.rs:445, from_slice at :195) and derives keys from EventChannelInstanceId::from_path at :208, so etcd/KV-store deployments mixing v1.2 and current binaries remain incompatible for event channels. If mixed-version rollout is only supported on Kubernetes this is fine; otherwise the shim is incomplete.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


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()),
);
Comment on lines +105 to +113

@devin-ai-integration devin-ai-integration Bot Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 Endpoint-scoped channels are downgraded to component identity, losing endpoint distinction for v1.2 readers

add_legacy_event_channel_fields maps every scope onto only namespace/component, so an EventScope::Endpoint channel becomes indistinguishable from a component-scoped one in the v1.2 wire shape (EventScope::component() returns the endpoint's component, lib/runtime/src/discovery/mod.rs:264-270). Two endpoint-scoped channels of the same component/topic on one worker share the same instance_id, so a v1.2 reader sees two map entries with identical legacy identity (namespace, component, topic, instance_id) but different transports; v1.2 keyed its own map by EventChannelInstanceId::to_path() (ns/comp/topic/id) and its subscribers group by discovery instance, so one publisher may be shadowed. Worth confirming whether the current runtime ever registers two endpoint-scoped channels with the same topic under one component, since that is the only scenario in which the downgrade is ambiguous. The reverse direction is also inherently lossy: upgraded v1.2 records always become EventScope::Component, so a current subscriber querying with EventChannelQuery::endpoint_topic(...) will not match a v1.2 publisher's record even for the same logical channel.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

}

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:
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand All @@ -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,
}]
);
}
}
2 changes: 1 addition & 1 deletion lib/runtime/src/discovery/kube/daemon.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,7 +291,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 {
Expand Down
Loading