Skip to content

perf(runtime): clone only the selected instance during dispatch #14545

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

Closed
jthomson04 wants to merge 2 commits into from
Closed

perf(runtime): clone only the selected instance during dispatch #14545

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fa68d17
perf(runtime): clone only the selected instance during dispatch
jthomson04 Sep 9, 2026
37bf2ae
test(runtime): trim redundant instance lookup checks
jthomson04 Sep 9, 2026
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
55 changes: 55 additions & 0 deletions lib/runtime/src/component/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -577,6 +577,14 @@ impl Client {
self.instance_source.borrow().clone()
}

pub(crate) fn instance_by_id(&self, instance_id: u64) -> Option<Instance> {
self.instance_source
.borrow()
.iter()
.find(|instance| instance.id() == instance_id)
.cloned()
}

pub fn instance_ids(&self) -> Vec<u64> {
self.instances().into_iter().map(|ep| ep.id()).collect()
}
Expand Down Expand Up @@ -978,6 +986,53 @@ mod tests {
use super::*;
use crate::{DistributedRuntime, Runtime, distributed::DistributedConfig};

#[tokio::test]
async fn test_instance_by_id_returns_owned_current_instance() {
use crate::component::TransportType;

let rt = Runtime::from_current().unwrap();
let drt = DistributedRuntime::new(rt.clone(), DistributedConfig::process_local())
.await
.unwrap();
let endpoint = drt
.namespace("test_instance_lookup".to_string())
.unwrap()
.component("test_component".to_string())
.unwrap()
.endpoint("test_endpoint".to_string());
let mut client = endpoint.client().await.unwrap();
let instances: Vec<_> = (1..=2)
.map(|instance_id| Instance {
namespace: "test_instance_lookup".to_string(),
component: "test_component".to_string(),
endpoint: "test_endpoint".to_string(),
instance_id,
transport: TransportType::Tcp(format!("127.0.0.1:{}", 9000 + instance_id)),
device_type: None,
request_plane_codec: None,
})
.collect();
let (tx, rx) = tokio::sync::watch::channel(instances.clone());
client.instance_source = Arc::new(rx);

assert_eq!(client.instance_by_id(1), Some(instances[0].clone()));
let selected = client.instance_by_id(2).unwrap();
assert_eq!(selected, instances[1]);
assert!(client.instance_by_id(3).is_none());

@dynamo-review-agent dynamo-review-agent Bot Sep 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

The never-present ID check is an extra missing-ID parameter that protects no distinct observable contract. instance_by_id only searches the current watch snapshot, so the later client.instance_by_id(2).is_none() after sending a snapshot without id 2 exercises the same no-match behavior while also preserving the current-snapshot removal regression; the id 2 equality assertion still protects selecting the requested non-first instance.

🤖 AI Fix

Remove the pre-update client.instance_by_id(3).is_none() assertion and keep the post-removal missing-ID assertion.


let mut updated = selected.clone();
updated.transport = TransportType::Nats("updated.subject".to_string());
tx.send(vec![instances[0].clone(), updated.clone()])
.unwrap();
assert_eq!(client.instance_by_id(2), Some(updated));

tx.send(vec![instances[0].clone()]).unwrap();
assert!(client.instance_by_id(2).is_none());
assert_eq!(client.instance_by_id(1), Some(instances[0].clone()));

rt.shutdown();
}

async fn wait_for_discovery_event(
receiver: &mut DiscoveryEventReceiver,
predicate: impl Fn(&DiscoveryEvent) -> bool,
Expand Down
22 changes: 9 additions & 13 deletions lib/runtime/src/pipeline/network/egress/push_router.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1840,19 +1840,15 @@ where
use crate::component::TransportType;

let lookup = |id: u64| {
self.client
.instances()
.iter()
.find(|i| i.instance_id == id)
.map(|instance| {
let (addr, kind) = match &instance.transport {
TransportType::Tcp(tcp_endpoint) => {
(tcp_endpoint.clone(), "transport.tcp.request")
}
TransportType::Nats(subject) => (subject.clone(), "transport.nats.request"),
};
(addr, kind, instance.clone())
})
self.client.instance_by_id(id).map(|instance| {
let (addr, kind) = match &instance.transport {
TransportType::Tcp(tcp_endpoint) => {
(tcp_endpoint.clone(), "transport.tcp.request")
}
TransportType::Nats(subject) => (subject.clone(), "transport.nats.request"),
};
(addr, kind, instance)
})
};

if let Some((addr, kind, inst)) = lookup(instance_id) {
Expand Down
Loading