diff --git a/Cargo.toml b/Cargo.toml index 01a9f0d0a..527271620 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -57,7 +57,7 @@ zenoh = { version = "1.9.0", default-features = false, features = [ "transport_tcp", "transport_serial", ] } -zenoh-ext = { version = "1.9.0", features = ["unstable"] } +zenoh-ext = { version = "1.9.0", default-features = false, features = ["unstable"] } zenoh-buffers = { version = "1.9.0" } # CLI diff --git a/crates/hiroz-protocol/Cargo.toml b/crates/hiroz-protocol/Cargo.toml index d89f77cea..397788dd6 100644 --- a/crates/hiroz-protocol/Cargo.toml +++ b/crates/hiroz-protocol/Cargo.toml @@ -12,7 +12,7 @@ categories = ["network-programming", "science::robotics"] [features] default = ["std", "rmw-zenoh"] -std = ["zenoh/default"] +std = [] rmw-zenoh = [] no-type-hash = [] # ROS 2 Humble doesn't support type hashing diff --git a/crates/hiroz-protocol/src/format/rmw_zenoh.rs b/crates/hiroz-protocol/src/format/rmw_zenoh.rs index 3b6d45754..80385de7c 100644 --- a/crates/hiroz-protocol/src/format/rmw_zenoh.rs +++ b/crates/hiroz-protocol/src/format/rmw_zenoh.rs @@ -318,6 +318,22 @@ mod tests { assert_eq!(decoded.durability, QosDurability::TransientLocal); } + #[test] + fn parse_liveliness_with_verbatim_rmw_system_default_qos() { + let key: KeyExpr<'_> = "@ros2_lv/123/1234567890abcdef1234567890abcdef/1/2/MP/%/%/res_gateway/%res%statuslight%autonomy/frost_msgs%msg%StatuslightRpdo1/RIHS01_0000000000000000000000000000000000000000000000000000000000000000/::,:,:,:,,:" + .try_into() + .unwrap(); + + let Entity::Endpoint(endpoint) = RmwZenohFormatter::parse_liveliness(&key).unwrap() else { + panic!("expected endpoint entity"); + }; + + assert_eq!(endpoint.node.unwrap().domain_id, 123); + assert_eq!(endpoint.kind, EndpointKind::Publisher); + assert_eq!(endpoint.topic, "/res/statuslight/autonomy"); + assert_eq!(endpoint.qos, QosProfile::default()); + } + /// Test topic key expression format matches rmw_zenoh. /// /// rmw_zenoh format: `///` diff --git a/crates/hiroz-protocol/src/qos.rs b/crates/hiroz-protocol/src/qos.rs index 68a2bfb67..c1579329d 100644 --- a/crates/hiroz-protocol/src/qos.rs +++ b/crates/hiroz-protocol/src/qos.rs @@ -77,7 +77,7 @@ impl QosProfile { // Parse reliability (RMW values: 1=Reliable, 2=BestEffort) let reliability = match fields[0] { - "" => default_qos.reliability, + "" | "0" => default_qos.reliability, "1" => QosReliability::Reliable, "2" => QosReliability::BestEffort, _ => return Err(QosDecodeError::InvalidReliability), @@ -85,28 +85,45 @@ impl QosProfile { // Parse durability (RMW values: 1=TransientLocal, 2=Volatile) let durability = match fields[1] { - "" => default_qos.durability, + "" | "0" => default_qos.durability, "1" => QosDurability::TransientLocal, "2" => QosDurability::Volatile, _ => return Err(QosDecodeError::InvalidDurability), }; - // Parse history: , - let history_parts: alloc::vec::Vec<&str> = fields[2].split(',').collect(); - if history_parts.len() < 2 { - return Err(QosDecodeError::InvalidHistory); - } + // Parse history: ,. rmw_zenoh_cpp omits QoS sub-fields + // whose value is SYSTEM_DEFAULT, so the history field can be just `,`. + let history = match fields[2] { + "," => default_qos.history, + // An omitted history field is only meaningful in the complete + // six-field wire representation. Keep rejecting truncated `::`. + "" if fields.len() >= 6 => default_qos.history, + encoded => { + let (kind, encoded_depth) = encoded + .split_once(',') + .ok_or(QosDecodeError::InvalidHistory)?; - let history = match history_parts[0] { - "" | "1" => { - // KeepLast - parse depth - let depth = history_parts[1] - .parse::() - .map_err(|_| QosDecodeError::InvalidHistory)?; - QosHistory::KeepLast(depth) + match kind { + "" | "0" | "1" => { + let depth = if encoded_depth.is_empty() { + default_qos.history.depth() + } else { + encoded_depth + .parse::() + .map_err(|_| QosDecodeError::InvalidHistory)? + }; + // A zero depth represents an unspecified/default depth + // at the ROS boundary; KeepLast(0) is not useful. + QosHistory::KeepLast(if depth == 0 { + default_qos.history.depth() + } else { + depth + }) + } + "2" => QosHistory::KeepAll, + _ => return Err(QosDecodeError::InvalidHistory), + } } - "2" => QosHistory::KeepAll, - _ => return Err(QosDecodeError::InvalidHistory), }; Ok(QosProfile { @@ -182,3 +199,77 @@ impl Display for QosDecodeError { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decode_rmw_compact_qos_corpus() { + let cases = [ + ("::,:,:,:,,", QosProfile::default()), + (":::,:,:,,", QosProfile::default()), + ("::1,:,:,:,,", QosProfile::default()), + ("::,10:,:,:,,", QosProfile::default()), + ( + "::2,:,:,:,,", + QosProfile { + history: QosHistory::KeepAll, + ..QosProfile::default() + }, + ), + ( + "1:1:,5:,:,:,,", + QosProfile { + durability: QosDurability::TransientLocal, + history: QosHistory::KeepLast(5), + ..QosProfile::default() + }, + ), + ( + "2::,1:,:,:,,", + QosProfile { + reliability: QosReliability::BestEffort, + history: QosHistory::KeepLast(1), + ..QosProfile::default() + }, + ), + ("0:0:0,0:,:,:,,", QosProfile::default()), + ]; + + for (encoded, expected) in cases { + assert_eq!(QosProfile::decode(encoded), Ok(expected), "{encoded}"); + } + } + + #[test] + fn qos_round_trip() { + let profiles = [ + QosProfile::default(), + QosProfile { + reliability: QosReliability::BestEffort, + durability: QosDurability::TransientLocal, + history: QosHistory::KeepLast(5), + }, + QosProfile { + history: QosHistory::KeepAll, + ..QosProfile::default() + }, + ]; + + for profile in profiles { + assert_eq!(QosProfile::decode(&profile.encode()), Ok(profile)); + } + } + + #[test] + fn reject_invalid_history() { + for encoded in ["::3,1:,:,:,,", "::1,-1:,:,:,,", "::"] { + assert_eq!( + QosProfile::decode(encoded), + Err(QosDecodeError::InvalidHistory), + "{encoded}" + ); + } + } +} diff --git a/crates/hiroz/src/dynamic/type_description_service.rs b/crates/hiroz/src/dynamic/type_description_service.rs index f863c09b3..4de8a6fb2 100644 --- a/crates/hiroz/src/dynamic/type_description_service.rs +++ b/crates/hiroz/src/dynamic/type_description_service.rs @@ -435,7 +435,11 @@ pub struct TypeDescriptionService { } impl TypeDescriptionService { - /// Create a new TypeDescriptionService. + /// Create a domain-0 TypeDescriptionService using the default key-expression format. + /// + /// Nodes created by [`ZNodeBuilder`](crate::node::ZNodeBuilder) inherit their context via + /// [`Self::new_with_node`]. Use that constructor when creating this service manually for a + /// non-default domain or key-expression format. /// /// # Arguments /// @@ -455,6 +459,34 @@ impl TypeDescriptionService { node_id: usize, counter: &crate::context::GlobalCounter, clock: &crate::time::ZClock, + ) -> ZResult { + let node = crate::entity::NodeEntity::new( + 0, + session.zid(), + node_id, + node_name.to_string(), + namespace.to_string(), + String::new(), + ); + Self::new_with_node( + session, + node, + counter, + clock, + hiroz_protocol::KeyExprFormat::default(), + ) + } + + /// Create a TypeDescriptionService from an existing node identity and key-expression format. + /// + /// `node.z_id` must identify `session`. Passing the owning node's entity avoids silently + /// resetting its domain, namespace, enclave, or other discovery identity. + pub fn new_with_node( + session: Arc, + node: crate::entity::NodeEntity, + counter: &crate::context::GlobalCounter, + clock: &crate::time::ZClock, + keyexpr_format: hiroz_protocol::KeyExprFormat, ) -> ZResult { let schemas: Arc>> = Arc::new(RwLock::new(HashMap::new())); @@ -464,19 +496,12 @@ impl TypeDescriptionService { // which expands to /{namespace}/{node_name}/get_type_description let service_name = "~get_type_description"; - // Create the node entity for the service - let node_entity = crate::entity::NodeEntity::new( - 0, // domain_id - session.zid(), - node_id, - node_name.to_string(), - namespace.to_string(), - String::new(), // enclave (empty, normalized to "%" in liveliness token) - ); + let node_name = node.name.clone(); + let namespace = node.namespace.clone(); let entity = crate::entity::EndpointEntity { id: counter.increment(), - node: Some(node_entity), + node: Some(node), kind: crate::entity::EndpointKind::Service, topic: service_name.to_string(), type_info: Some(GetTypeDescription::service_type_info()), @@ -488,7 +513,7 @@ impl TypeDescriptionService { entity, session, clock: clock.clone(), - keyexpr_format: hiroz_protocol::KeyExprFormat::default(), + keyexpr_format, _phantom_data: Default::default(), }; diff --git a/crates/hiroz/src/node.rs b/crates/hiroz/src/node.rs index e640a2191..0f0fd86c9 100644 --- a/crates/hiroz/src/node.rs +++ b/crates/hiroz/src/node.rs @@ -241,13 +241,12 @@ impl Builder for ZNodeBuilder { // Create type description service if enabled let type_desc_service = if self.enable_type_desc_service { debug!("[NOD] Creating type description service"); - let service = TypeDescriptionService::new( + let service = TypeDescriptionService::new_with_node( self.session.clone(), - &self.name, - &self.namespace, - id, + node.clone(), &self.counter, &self.clock, + self.keyexpr_format.clone(), )?; info!("[NOD] TypeDescriptionService created (callback mode)"); @@ -263,6 +262,8 @@ impl Builder for ZNodeBuilder { let service = ParameterService::new(ParameterServiceConfig { session: self.session.clone(), graph: self.graph.clone(), + domain_id: self.domain_id, + keyexpr_format: self.keyexpr_format.clone(), node_name: &self.name, namespace: &self.namespace, node_id: id, diff --git a/crates/hiroz/src/parameter/service.rs b/crates/hiroz/src/parameter/service.rs index 15db88d81..8ebc2c236 100644 --- a/crates/hiroz/src/parameter/service.rs +++ b/crates/hiroz/src/parameter/service.rs @@ -40,6 +40,8 @@ type BoxedServer = Arc; pub(crate) struct ParameterServiceConfig<'a> { pub session: Arc, pub graph: Arc, + pub domain_id: usize, + pub keyexpr_format: hiroz_protocol::KeyExprFormat, pub node_name: &'a str, pub namespace: &'a str, pub node_id: usize, @@ -284,6 +286,8 @@ impl ParameterService { let ParameterServiceConfig { session, graph, + domain_id, + keyexpr_format, node_name, namespace, node_id, @@ -297,7 +301,7 @@ impl ParameterService { wire_types::register_parameter_schemas(tds); } let node_entity = NodeEntity::new( - 0, + domain_id, session.zid(), node_id, node_name.to_string(), @@ -323,8 +327,6 @@ impl ParameterService { qos: Default::default(), }; - let ke_format = hiroz_protocol::KeyExprFormat::default(); - // ── /parameter_events publisher ─────────────────────────────────────── let pub_entity = EndpointEntity { id: counter.increment(), @@ -351,7 +353,7 @@ impl ParameterService { clock: clock.clone(), with_attachment: true, shm_config: None, - keyexpr_format: ke_format.clone(), + keyexpr_format: keyexpr_format.clone(), dyn_schema: None, encoding: None, _phantom_data: Default::default(), @@ -380,7 +382,7 @@ impl ParameterService { entity, session: session.clone(), clock: clock.clone(), - keyexpr_format: ke_format.clone(), + keyexpr_format: keyexpr_format.clone(), _phantom_data: Default::default(), }; builder.build_with_callback(move |query| { @@ -400,7 +402,7 @@ impl ParameterService { entity, session: session.clone(), clock: clock.clone(), - keyexpr_format: ke_format.clone(), + keyexpr_format: keyexpr_format.clone(), _phantom_data: Default::default(), }; builder.build_with_callback(move |query| { @@ -420,7 +422,7 @@ impl ParameterService { entity, session: session.clone(), clock: clock.clone(), - keyexpr_format: ke_format.clone(), + keyexpr_format: keyexpr_format.clone(), _phantom_data: Default::default(), }; builder.build_with_callback(move |query| { @@ -440,7 +442,7 @@ impl ParameterService { entity, session: session.clone(), clock: clock.clone(), - keyexpr_format: ke_format.clone(), + keyexpr_format: keyexpr_format.clone(), _phantom_data: Default::default(), }; builder.build_with_callback(move |query| { @@ -460,7 +462,7 @@ impl ParameterService { entity, session: session.clone(), clock: clock.clone(), - keyexpr_format: ke_format.clone(), + keyexpr_format: keyexpr_format.clone(), _phantom_data: Default::default(), }; builder.build_with_callback(move |query| { @@ -480,7 +482,7 @@ impl ParameterService { entity, session: session.clone(), clock: clock.clone(), - keyexpr_format: ke_format.clone(), + keyexpr_format, _phantom_data: Default::default(), }; builder.build_with_callback(move |query| { diff --git a/crates/hiroz/tests/graph.rs b/crates/hiroz/tests/graph.rs index 518cb33ac..2d44e08eb 100644 --- a/crates/hiroz/tests/graph.rs +++ b/crates/hiroz/tests/graph.rs @@ -7,7 +7,7 @@ //! - Node discovery and information //! - Service availability checking -use std::time::Duration; +use std::{sync::Arc, time::Duration}; use hiroz::{ Builder, Result, @@ -15,6 +15,95 @@ use hiroz::{ entity::{EndpointKind, NodeKey}, }; use hiroz_msgs::{example_interfaces::srv::AddTwoInts, std_msgs::String as RosString}; +use hiroz_protocol::{ + EndpointEntity, Entity, KeyExprFormat, KeyExprFormatter, KeyExprFormatterAdapter, NodeEntity, + RmwZenohFormatter, + entity::{LivelinessKE, TopicKE}, + qos::QosProfile, +}; +use zenoh::{Wait, config::WhatAmI}; + +#[derive(Debug)] +struct TestKeyExprFormatter; + +impl TestKeyExprFormatter { + fn from_rmw_liveliness(key: LivelinessKE) -> hiroz::Result { + Ok(LivelinessKE::new( + key.as_str() + .replacen("@ros2_lv", Self::ADMIN_SPACE, 1) + .try_into()?, + )) + } +} + +impl KeyExprFormatter for TestKeyExprFormatter { + const ESCAPE_CHAR: char = '%'; + const ADMIN_SPACE: &'static str = "@hiroz_test_lv"; + + fn topic_key_expr(entity: &EndpointEntity) -> hiroz::Result { + let rmw = RmwZenohFormatter::topic_key_expr(entity)?; + Ok(TopicKE::new( + format!("hiroz_test/{}", rmw.as_str()).try_into()?, + )) + } + + fn liveliness_key_expr( + entity: &EndpointEntity, + zid: &zenoh::session::ZenohId, + ) -> hiroz::Result { + Self::from_rmw_liveliness(RmwZenohFormatter::liveliness_key_expr(entity, zid)?) + } + + fn node_liveliness_key_expr(entity: &NodeEntity) -> hiroz::Result { + Self::from_rmw_liveliness(RmwZenohFormatter::node_liveliness_key_expr(entity)?) + } + + fn parse_liveliness(key: &zenoh::key_expr::KeyExpr) -> hiroz::Result { + let rmw: zenoh::key_expr::KeyExpr<'static> = key + .as_str() + .replacen(Self::ADMIN_SPACE, "@ros2_lv", 1) + .try_into()?; + RmwZenohFormatter::parse_liveliness(&rmw) + } + + fn encode_qos(qos: &QosProfile, keyless: bool) -> String { + RmwZenohFormatter::encode_qos(qos, keyless) + } + + fn decode_qos(encoded: &str) -> hiroz::Result<(bool, QosProfile)> { + RmwZenohFormatter::decode_qos(encoded) + } +} + +struct DomainTestRouter { + endpoint: String, + _session: zenoh::Session, +} + +impl DomainTestRouter { + fn new() -> Self { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind test port"); + let port = listener.local_addr().expect("test address").port(); + drop(listener); + + let endpoint = format!("tcp/127.0.0.1:{port}"); + let mut config = zenoh::Config::default(); + config.set_mode(Some(WhatAmI::Router)).unwrap(); + config + .insert_json5("listen/endpoints", &format!("[\"{endpoint}\"]")) + .unwrap(); + config + .insert_json5("scouting/multicast/enabled", "false") + .unwrap(); + let session = zenoh::open(config).wait().expect("open test router"); + + Self { + endpoint, + _session: session, + } + } +} + /// Helper to create a test context and node async fn setup_test_node( node_name: &str, @@ -74,6 +163,125 @@ async fn wait_for_subscribers( mod tests { use super::*; + #[tokio::test(flavor = "multi_thread")] + async fn built_in_services_inherit_context_domain() -> Result<()> { + const DOMAIN_ID: usize = 123; + let router = DomainTestRouter::new(); + let observer_ctx = ZContextBuilder::default() + .with_domain_id(DOMAIN_ID) + .disable_multicast_scouting() + .with_connect_endpoints([router.endpoint.as_str()]) + .with_mode("client") + .build()?; + let observer = observer_ctx + .create_node("domain_123_observer") + .without_parameters() + .build()?; + let producer_ctx = ZContextBuilder::default() + .with_domain_id(DOMAIN_ID) + .disable_multicast_scouting() + .with_connect_endpoints([router.endpoint.as_str()]) + .with_mode("client") + .build()?; + let _producer = producer_ctx + .create_node("domain_123_builtins") + .with_type_description_service() + .build()?; + let node_key: NodeKey = (String::new(), "domain_123_builtins".to_string()); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let services = loop { + let services = observer + .graph() + .get_entities_by_node(EndpointKind::Service, node_key.clone()); + if services.len() >= 7 || tokio::time::Instant::now() >= deadline { + break services; + } + tokio::time::sleep(Duration::from_millis(10)).await; + }; + + assert_eq!( + services.len(), + 7, + "six parameter services and get_type_description must be discoverable" + ); + assert!(services.iter().all(|endpoint| { + endpoint + .node + .as_ref() + .is_some_and(|owner| owner.domain_id == DOMAIN_ID) + })); + + let parameter_events = observer + .graph() + .get_entities_by_node(EndpointKind::Publisher, node_key); + assert!(parameter_events.iter().any(|endpoint| { + endpoint.topic == "/parameter_events" + && endpoint + .node + .as_ref() + .is_some_and(|owner| owner.domain_id == DOMAIN_ID) + })); + + Ok(()) + } + + #[tokio::test(flavor = "multi_thread")] + async fn built_in_services_inherit_custom_keyexpr_format() -> Result<()> { + let router = DomainTestRouter::new(); + let format = KeyExprFormat::Custom(Arc::new( + KeyExprFormatterAdapter::::new(), + )); + let observer_ctx = ZContextBuilder::default() + .keyexpr_format(format.clone()) + .disable_multicast_scouting() + .with_connect_endpoints([router.endpoint.as_str()]) + .with_mode("client") + .build()?; + let observer = observer_ctx + .create_node("custom_format_observer") + .without_parameters() + .build()?; + let producer_ctx = ZContextBuilder::default() + .keyexpr_format(format) + .disable_multicast_scouting() + .with_connect_endpoints([router.endpoint.as_str()]) + .with_mode("client") + .build()?; + let _producer = producer_ctx + .create_node("custom_format_builtins") + .with_type_description_service() + .build()?; + let node_key: NodeKey = (String::new(), "custom_format_builtins".to_string()); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let services = loop { + let services = observer + .graph() + .get_entities_by_node(EndpointKind::Service, node_key.clone()); + if services.len() >= 7 || tokio::time::Instant::now() >= deadline { + break services; + } + tokio::time::sleep(Duration::from_millis(10)).await; + }; + + assert_eq!( + services.len(), + 7, + "built-in services must use the context's custom liveliness format" + ); + assert!( + observer + .graph() + .get_entities_by_node(EndpointKind::Publisher, node_key) + .iter() + .any(|endpoint| endpoint.topic == "/parameter_events"), + "parameter-events publisher must use the context's custom liveliness format" + ); + + Ok(()) + } + /// Tests getting topic names and types from the graph #[tokio::test(flavor = "multi_thread")] async fn test_get_topic_names_and_types() -> Result<()> { diff --git a/docs/patches/0001-Fix-rmw-graph-bootstrap-with-default-QoS.patch b/docs/patches/0001-Fix-rmw-graph-bootstrap-with-default-QoS.patch new file mode 100644 index 000000000..79cc1839e --- /dev/null +++ b/docs/patches/0001-Fix-rmw-graph-bootstrap-with-default-QoS.patch @@ -0,0 +1,341 @@ +From 4171acd0d438f2acfc3e041c2efe14d90a9e191e Mon Sep 17 00:00:00 2001 +From: Lukas Rieger +Date: Sat, 29 Aug 2026 17:40:34 +0200 +Subject: [PATCH] Fix rmw graph bootstrap with default QoS + +ARCHIVED DEVELOPMENT PATCH -- DO NOT APPLY AS A WHOLE. + +This file preserves the exact source diff from commit 4171acd for incident and +design history. The QoS portion correctly accepts rmw_zenoh's omitted/default +history depth. The asynchronous graph-bootstrap portion was rejected after +review because snapshot entities bypass graph notifications, construction can +return before discovery begins, failures are hidden, replies can be truncated, +and the detached worker has no lifecycle owner. + +The safe follow-up keeps and hardens the QoS parser change, restores the graph +and context implementation to the recorded base, and fixes domain/key-expression +propagation for built-in services. See docs/patches/README.md for details. + +--- + crates/hiroz-protocol/src/qos.rs | 42 +++++++++- + crates/hiroz/src/context.rs | 12 ++- + crates/hiroz/src/graph.rs | 128 ++++++++++++++++++++++++++----- + 3 files changed, 156 insertions(+), 26 deletions(-) + +diff --git a/crates/hiroz-protocol/src/qos.rs b/crates/hiroz-protocol/src/qos.rs +index 68a2bfb..2569e6b 100644 +--- a/crates/hiroz-protocol/src/qos.rs ++++ b/crates/hiroz-protocol/src/qos.rs +@@ -99,10 +99,17 @@ impl QosProfile { + + let history = match history_parts[0] { + "" | "1" => { +- // KeepLast - parse depth +- let depth = history_parts[1] +- .parse::() +- .map_err(|_| QosDecodeError::InvalidHistory)?; ++ // rmw_zenoh_cpp leaves both history kind and depth empty for ++ // SYSTEM_DEFAULT QoS (the resulting field is just `,`). ++ // Treat an omitted depth as our ROS-compatible default rather ++ // than rejecting the entire liveliness token. ++ let depth = if history_parts[1].is_empty() { ++ default_qos.history.depth() ++ } else { ++ history_parts[1] ++ .parse::() ++ .map_err(|_| QosDecodeError::InvalidHistory)? ++ }; + QosHistory::KeepLast(depth) + } + "2" => QosHistory::KeepAll, +@@ -182,3 +189,30 @@ impl Display for QosDecodeError { + } + } + } ++ ++#[cfg(test)] ++mod tests { ++ use super::*; ++ ++ #[test] ++ fn decode_rmw_system_default_history() { ++ let decoded = QosProfile::decode("::,:,:,:,,").unwrap(); ++ ++ assert_eq!(decoded, QosProfile::default()); ++ } ++ ++ #[test] ++ fn decode_explicit_keep_last_with_default_depth() { ++ let decoded = QosProfile::decode("::1,:,:,:,,").unwrap(); ++ ++ assert_eq!(decoded.history, QosHistory::KeepLast(10)); ++ } ++ ++ #[test] ++ fn reject_non_numeric_history_depth() { ++ assert_eq!( ++ QosProfile::decode("::1,not-a-number:,:,:,,"), ++ Err(QosDecodeError::InvalidHistory) ++ ); ++ } ++} +diff --git a/crates/hiroz/src/context.rs b/crates/hiroz/src/context.rs +index ba6d02f..18314d7 100644 +--- a/crates/hiroz/src/context.rs ++++ b/crates/hiroz/src/context.rs +@@ -1,6 +1,7 @@ + use std::{ + collections::HashMap, + sync::{Arc, atomic::AtomicUsize}, ++ time::Duration, + }; + + use tracing::{debug, warn}; +@@ -84,6 +85,7 @@ pub struct ZContextBuilder { + shm_config: Option>, + keyexpr_format: hiroz_protocol::KeyExprFormat, + clock: Option, ++ graph_bootstrap_delay: Duration, + } + + impl ZContextBuilder { +@@ -93,6 +95,13 @@ impl ZContextBuilder { + self + } + ++ /// Delay remote graph discovery so latency-sensitive local entities can be ++ /// declared before a large router graph is synchronized. ++ pub fn with_graph_bootstrap_delay(mut self, delay: Duration) -> Self { ++ self.graph_bootstrap_delay = delay; ++ self ++ } ++ + /// Set the default namespace inherited by nodes created from this context. + pub fn with_namespace(mut self, namespace: impl AsRef) -> Self { + self.namespace = normalize_node_namespace(namespace.as_ref()); +@@ -551,10 +560,11 @@ impl Builder for ZContextBuilder { + } + + let domain_id = builder.domain_id; +- let graph = Arc::new(Graph::new( ++ let graph = Arc::new(Graph::new_with_bootstrap_delay( + &session, + domain_id, + builder.keyexpr_format.clone(), ++ builder.graph_bootstrap_delay, + )?); + + Ok(ZContext { +diff --git a/crates/hiroz/src/graph.rs b/crates/hiroz/src/graph.rs +index 1e4d8c5..5209cf7 100644 +--- a/crates/hiroz/src/graph.rs ++++ b/crates/hiroz/src/graph.rs +@@ -4,7 +4,7 @@ use slab::Slab; + use std::{ + collections::{HashMap, HashSet}, + sync::{Arc, Condvar, Mutex as StdMutex, Weak}, +- time::{Duration, SystemTime}, ++ time::{Duration, Instant, SystemTime}, + }; + use tokio::sync::Notify; + use tracing::debug; +@@ -435,7 +435,7 @@ pub struct Graph { + /// The liveliness callback holds `data` first, then releases it, then acquires this + /// mutex — so both locks are never held simultaneously. + pub change_signal: Arc<(StdMutex<()>, Condvar)>, +- _subscriber: Subscriber<()>, ++ _subscriber: Arc>>>, + } + + impl std::fmt::Debug for Graph { +@@ -455,12 +455,25 @@ impl Graph { + session: &Session, + domain_id: usize, + format: hiroz_protocol::KeyExprFormat, ++ ) -> Result { ++ Self::new_with_bootstrap_delay(session, domain_id, format, Duration::ZERO) ++ } ++ ++ pub fn new_with_bootstrap_delay( ++ session: &Session, ++ domain_id: usize, ++ format: hiroz_protocol::KeyExprFormat, ++ bootstrap_delay: Duration, + ) -> Result { + let liveliness_pattern = format.liveliness_pattern(domain_id); + +- Self::new_with_pattern(session, domain_id, liveliness_pattern, move |ke| { +- format.parse_liveliness(ke) +- }) ++ Self::new_with_pattern_and_bootstrap_delay( ++ session, ++ domain_id, ++ liveliness_pattern, ++ bootstrap_delay, ++ move |ke| format.parse_liveliness(ke), ++ ) + } + + pub(crate) async fn wait_until(&self, timeout: Duration, predicate: F) -> bool +@@ -573,9 +586,28 @@ impl Graph { + /// # Pattern + /// * RmwZenoh: `@ros2_lv/{domain_id}/**` + pub fn new_with_pattern( ++ session: &Session, ++ domain_id: usize, ++ liveliness_pattern: String, ++ parser: F, ++ ) -> Result ++ where ++ F: Fn(&zenoh::key_expr::KeyExpr) -> Result + Send + Sync + 'static, ++ { ++ Self::new_with_pattern_and_bootstrap_delay( ++ session, ++ domain_id, ++ liveliness_pattern, ++ Duration::ZERO, ++ parser, ++ ) ++ } ++ ++ fn new_with_pattern_and_bootstrap_delay( + session: &Session, + _domain_id: usize, + liveliness_pattern: String, ++ bootstrap_delay: Duration, + parser: F, + ) -> Result + where +@@ -594,11 +626,30 @@ impl Graph { + let c_zid = zid; + let c_liveliness_pattern = liveliness_pattern.clone(); + let callback_parser = parser_arc.clone(); +- tracing::debug!("Creating liveliness subscriber for {}", liveliness_pattern); +- let sub = session ++ let query_graph_data = graph_data.clone(); ++ let query_parser = parser_arc.clone(); ++ let subscriber_slot = Arc::new(Mutex::new(None)); ++ let c_subscriber_slot = subscriber_slot.clone(); ++ let bootstrap_session = session.clone(); ++ let (bootstrap_ready_tx, bootstrap_ready_rx) = std::sync::mpsc::sync_channel(1); ++ ++ // Zenoh can hold its session-state lock for an unbounded time while a ++ // newly connected client ingests a large ROS graph. Bootstrap discovery ++ // in the background so context construction and local entity declaration ++ // are never held hostage by that remote graph synchronization. ++ let bootstrap_result = std::thread::Builder::new() ++ .name("hiroz-graph-bootstrap".to_owned()) ++ .spawn(move || { ++ const LIVELINESS_QUERY_TIMEOUT: Duration = Duration::from_secs(3); ++ ++ std::thread::sleep(bootstrap_delay); ++ tracing::debug!( ++ "Creating liveliness subscriber for {}", ++ c_liveliness_pattern ++ ); ++ let sub = match bootstrap_session + .liveliness() +- .declare_subscriber(&liveliness_pattern) +- .history(true) ++ .declare_subscriber(&c_liveliness_pattern) + .callback(move |sample| { + let key_expr = sample.key_expr().to_owned(); + let ke = LivelinessKE(key_expr.clone()); +@@ -684,16 +735,34 @@ impl Graph { + // so the callback must never hold data while acquiring change_signal.0. + c_change_signal.1.notify_all(); + }) +- .wait()?; +- +- // Query existing liveliness tokens from all connected sessions +- // This is crucial for cross-context discovery where entities from other sessions +- // were created before this session started +- let replies = session ++ .wait() ++ { ++ Ok(sub) => sub, ++ Err(error) => { ++ tracing::warn!("Failed to initialize graph subscriber: {error}"); ++ return; ++ } ++ }; ++ *c_subscriber_slot.lock() = Some(sub); ++ let _ = bootstrap_ready_tx.send(()); ++ ++ // Query existing liveliness tokens from all connected sessions. The live ++ // subscriber is declared first so entities appearing during this query ++ // cannot be missed. Do not also request subscriber history here: on a ++ // large ROS graph that makes declaration itself an unbounded blocking ++ // bootstrap, before the bounded query below can apply its timeout. ++ let replies = match bootstrap_session + .liveliness() + .get(&c_liveliness_pattern) +- .timeout(std::time::Duration::from_secs(3)) +- .wait()?; ++ .timeout(LIVELINESS_QUERY_TIMEOUT) ++ .wait() ++ { ++ Ok(replies) => replies, ++ Err(error) => { ++ tracing::warn!("Failed to query initial graph snapshot: {error}"); ++ return; ++ } ++ }; + + // Process all replies and add them to the graph. + // Filter current-session entities: add_local_entity() already inserted them, so +@@ -702,13 +771,21 @@ impl Graph { + // Endpoints without node identity carry no z_id and are never filtered. + let mut reply_count = 0; + let mut filtered_count = 0; +- while let Ok(reply) = replies.recv() { ++ // Zenoh's query timeout does not always close the reply channel when a ++ // router remains connected. Apply the same timeout to the receive side ++ // so graph construction cannot wait forever after processing the ++ // current snapshot. ++ let reply_deadline = Instant::now() + LIVELINESS_QUERY_TIMEOUT; ++ while Instant::now() < reply_deadline { ++ let Ok(Some(reply)) = replies.recv_deadline(reply_deadline) else { ++ break; ++ }; + reply_count += 1; + if let Ok(sample) = reply.into_result() { + let key_expr = sample.key_expr().to_owned(); + let ke = LivelinessKE(key_expr.clone()); + +- if let Ok(entity) = parser_arc(&key_expr) { ++ if let Ok(entity) = query_parser(&key_expr) { + let is_local = match &entity { + Entity::Node(node) => node.z_id == zid, + Entity::Endpoint(endpoint) => { +@@ -723,7 +800,7 @@ impl Graph { + } + + tracing::debug!("Graph: Caching cross-context entity: {}", key_expr.as_str()); +- graph_data.lock().insert(ke); ++ query_graph_data.lock().insert(ke); + } + } + tracing::debug!( +@@ -731,9 +808,18 @@ impl Graph { + reply_count, + filtered_count + ); ++ }); ++ ++ if let Err(error) = bootstrap_result { ++ return Err(format!("Failed to spawn graph bootstrap thread: {error}").into()); ++ } ++ // Preserve synchronous discovery for the usual small/local graph. On a ++ // congested router this short wait expires and local entity creation can ++ // proceed while remote discovery continues in the bootstrap thread. ++ let _ = bootstrap_ready_rx.recv_timeout(Duration::from_millis(100)); + + Ok(Self { +- _subscriber: sub, ++ _subscriber: subscriber_slot, + data: graph_data, + event_manager, + change_notify, +-- +2.54.0 diff --git a/docs/patches/README.md b/docs/patches/README.md new file mode 100644 index 000000000..f0e5f5cd0 --- /dev/null +++ b/docs/patches/README.md @@ -0,0 +1,30 @@ +# Archived development patches + +This directory preserves patches that are useful as design or incident records but are not +intended to be applied unchanged. + +## `0001-Fix-rmw-graph-bootstrap-with-default-QoS.patch` + +- Original commit: `4171acd` (`Fix rmw graph bootstrap with default QoS`) +- Based on: `c7bf42d87b08a82df35c9b716c85e5b0b6676781` +- Incident: endpoints using rmw_zenoh's omitted/system-default history depth were dropped while + parsing graph liveliness tokens on ROS domain 123. + +The patch is archived intact because it contains both the successful QoS parser fix and an +experimental asynchronous graph bootstrap. **Do not apply it as a whole.** Review found that the +graph portion can return an empty or partial graph before bootstrap begins, bypass graph-change +events for snapshot entities, hide initialization failures, truncate snapshots, and leave an +unowned background thread alive. + +The production-safe follow-up keeps the QoS correction separately and restores the synchronous, +history-enabled graph construction. The original graph experiment remains here so its approach, +motivation, and failure modes are not lost. + +For forensic use, inspect it with: + +```console +git apply --stat docs/patches/0001-Fix-rmw-graph-bootstrap-with-default-QoS.patch +git apply --check docs/patches/0001-Fix-rmw-graph-bootstrap-with-default-QoS.patch +``` + +Only apply it to a disposable branch created from the recorded base commit.