From 0bb218ff922d11bfda334ae65404a9f47cd5903f Mon Sep 17 00:00:00 2001 From: Jeff Hykin Date: Thu, 27 Aug 2026 16:32:53 -0700 Subject: [PATCH 1/3] fix(native): honor LCM_DEFAULT_URL in the rust LCM transport liblcm reads LCM_DEFAULT_URL, so every python module lands on the bus it names, but the rust transport hardcoded 239.255.76.67:7667. The two halves of a pipeline then sit on different buses and nothing is delivered, with no error on either side. This is also why no python-to-native LCM integration test can pass: dimos' conftest pins each pytest session to its own udpm port for isolation, which the native side never joined. --- native/rust/dimos-module/src/lcm.rs | 78 ++++++++++++++++++++++++++++- 1 file changed, 77 insertions(+), 1 deletion(-) diff --git a/native/rust/dimos-module/src/lcm.rs b/native/rust/dimos-module/src/lcm.rs index 5dbf0c3986..675727a604 100644 --- a/native/rust/dimos-module/src/lcm.rs +++ b/native/rust/dimos-module/src/lcm.rs @@ -36,9 +36,50 @@ pub struct LcmTransport { runtime: tokio::runtime::Handle, } +/// liblcm reads this, so python modules follow it and native ones have to as well or the +/// two halves of a pipeline end up on different buses. Format: `udpm://group:port?ttl=n`. +fn options_from_env() -> LcmOptions { + match std::env::var("LCM_DEFAULT_URL") { + Ok(url) => options_from_url(&url), + Err(_) => LcmOptions::default(), + } +} + +fn options_from_url(url: &str) -> LcmOptions { + let mut options = LcmOptions::default(); + let Some(rest) = url.strip_prefix("udpm://") else { + tracing::warn!( + url, + "LCM_DEFAULT_URL is not a udpm:// url; using the defaults" + ); + return options; + }; + let (address, query) = rest.split_once('?').unwrap_or((rest, "")); + if let Some((group, port)) = address.rsplit_once(':') { + match (group.parse(), port.parse()) { + (Ok(group), Ok(port)) => { + options.multicast_group = group; + options.port = port; + } + _ => tracing::warn!( + url, + "LCM_DEFAULT_URL has no parsable group:port; using the defaults" + ), + } + } + for (key, value) in query.split('&').filter_map(|pair| pair.split_once('=')) { + if key == "ttl" { + if let Ok(ttl) = value.parse() { + options.ttl = ttl; + } + } + } + options +} + impl LcmTransport { pub async fn new() -> io::Result { - Ok(Self::wrap(Lcm::new().await?)) + Self::with_options(options_from_env()).await } pub async fn with_options(opts: LcmOptions) -> io::Result { @@ -119,3 +160,38 @@ impl Transport for LcmTransport { } } } + +#[cfg(test)] +mod tests { + use super::options_from_url; + use std::net::Ipv4Addr; + + #[test] + fn reads_group_port_and_ttl() { + let options = options_from_url("udpm://239.255.76.67:7712?ttl=0"); + assert_eq!(options.multicast_group, Ipv4Addr::new(239, 255, 76, 67)); + assert_eq!(options.port, 7712); + assert_eq!(options.ttl, 0); + } + + #[test] + fn ttl_is_optional() { + let options = options_from_url("udpm://239.255.76.67:7712"); + assert_eq!(options.port, 7712); + assert_eq!(options.ttl, dimos_lcm::LcmOptions::default().ttl); + } + + #[test] + fn an_unusable_url_leaves_the_defaults() { + let defaults = dimos_lcm::LcmOptions::default(); + for url in [ + "tcp://127.0.0.1:7667", + "udpm://not-an-ip:7667", + "udpm://239.255.76.67", + ] { + let options = options_from_url(url); + assert_eq!(options.multicast_group, defaults.multicast_group, "{url}"); + assert_eq!(options.port, defaults.port, "{url}"); + } + } +} From 70787876c9a7c92a335432a5ac9e72198488c084 Mon Sep 17 00:00:00 2001 From: Jeff Hykin Date: Thu, 27 Aug 2026 17:13:46 -0700 Subject: [PATCH 2/3] fix(native): parse LCM_DEFAULT_URL with the url crate Hand-splitting on "://", "?", ":" and "&" reimplements a parser that is already a dependency and gets the edge cases wrong. --- Cargo.lock | 1 + native/rust/dimos-module/Cargo.toml | 1 + native/rust/dimos-module/src/lcm.rs | 40 ++++++++++++++++------------- 3 files changed, 24 insertions(+), 18 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index eec685b1d9..415196cd6f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -624,6 +624,7 @@ dependencies = [ "tracing", "tracing-subscriber", "tracing-test", + "url", "validator", "zenoh", ] diff --git a/native/rust/dimos-module/Cargo.toml b/native/rust/dimos-module/Cargo.toml index bfa69eb5fd..10c86c20a8 100644 --- a/native/rust/dimos-module/Cargo.toml +++ b/native/rust/dimos-module/Cargo.toml @@ -25,6 +25,7 @@ serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["json", "env-filter"] } +url = "2" validator = { version = "0.20", features = ["derive"] } # Default features pull in ring via QUIC/TLS, which breaks static musl cross-linking. zenoh = { version = "=1.9.0", default-features = false, features = ["transport_tcp", "transport_udp", "unstable", "shared-memory"] } diff --git a/native/rust/dimos-module/src/lcm.rs b/native/rust/dimos-module/src/lcm.rs index 675727a604..47ef251b07 100644 --- a/native/rust/dimos-module/src/lcm.rs +++ b/native/rust/dimos-module/src/lcm.rs @@ -19,6 +19,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use dimos_lcm::{Lcm, LcmOptions}; +use url::Url; use crate::transport::{Dispatch, Transport}; @@ -47,27 +48,29 @@ fn options_from_env() -> LcmOptions { fn options_from_url(url: &str) -> LcmOptions { let mut options = LcmOptions::default(); - let Some(rest) = url.strip_prefix("udpm://") else { - tracing::warn!( - url, - "LCM_DEFAULT_URL is not a udpm:// url; using the defaults" - ); - return options; - }; - let (address, query) = rest.split_once('?').unwrap_or((rest, "")); - if let Some((group, port)) = address.rsplit_once(':') { - match (group.parse(), port.parse()) { - (Ok(group), Ok(port)) => { - options.multicast_group = group; - options.port = port; - } - _ => tracing::warn!( + let parsed = match Url::parse(url) { + Ok(parsed) if parsed.scheme() == "udpm" => parsed, + _ => { + tracing::warn!( url, - "LCM_DEFAULT_URL has no parsable group:port; using the defaults" - ), + "LCM_DEFAULT_URL is not a udpm:// url; using the defaults" + ); + return options; + } + }; + // udpm is not a special scheme, so the host stays an opaque string. + let group = parsed.host_str().and_then(|host| host.parse().ok()); + match (group, parsed.port()) { + (Some(group), Some(port)) => { + options.multicast_group = group; + options.port = port; } + _ => tracing::warn!( + url, + "LCM_DEFAULT_URL has no parsable group:port; using the defaults" + ), } - for (key, value) in query.split('&').filter_map(|pair| pair.split_once('=')) { + for (key, value) in parsed.query_pairs() { if key == "ttl" { if let Ok(ttl) = value.parse() { options.ttl = ttl; @@ -188,6 +191,7 @@ mod tests { "tcp://127.0.0.1:7667", "udpm://not-an-ip:7667", "udpm://239.255.76.67", + "239.255.76.67:7667", ] { let options = options_from_url(url); assert_eq!(options.multicast_group, defaults.multicast_group, "{url}"); From df2bfdd5324861b901bc4d3a8ba5f611d476824a Mon Sep 17 00:00:00 2001 From: Andrew Lauer Date: Mon, 31 Aug 2026 16:15:28 -0700 Subject: [PATCH 3/3] Warn and return defaults --- native/rust/dimos-module/src/lcm.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/native/rust/dimos-module/src/lcm.rs b/native/rust/dimos-module/src/lcm.rs index 47ef251b07..6fe3838253 100644 --- a/native/rust/dimos-module/src/lcm.rs +++ b/native/rust/dimos-module/src/lcm.rs @@ -65,10 +65,13 @@ fn options_from_url(url: &str) -> LcmOptions { options.multicast_group = group; options.port = port; } - _ => tracing::warn!( - url, - "LCM_DEFAULT_URL has no parsable group:port; using the defaults" - ), + _ => { + tracing::warn!( + url, + "LCM_DEFAULT_URL has no parsable group:port; using the defaults" + ); + return options; + } } for (key, value) in parsed.query_pairs() { if key == "ttl" { @@ -189,13 +192,14 @@ mod tests { let defaults = dimos_lcm::LcmOptions::default(); for url in [ "tcp://127.0.0.1:7667", - "udpm://not-an-ip:7667", - "udpm://239.255.76.67", + "udpm://not-an-ip:7667?ttl=42", + "udpm://239.255.76.67?ttl=42", "239.255.76.67:7667", ] { let options = options_from_url(url); assert_eq!(options.multicast_group, defaults.multicast_group, "{url}"); assert_eq!(options.port, defaults.port, "{url}"); + assert_eq!(options.ttl, defaults.ttl, "{url}"); } } }