diff --git a/docs/config/discovery-options.rst b/docs/config/discovery-options.rst index c47613dc0..d928e8ef6 100644 --- a/docs/config/discovery-options.rst +++ b/docs/config/discovery-options.rst @@ -51,7 +51,8 @@ In runtime mode, the gateway maps the ROS 2 graph to SOVD entities as follows: - **Components** - a single host-level Component is created from ``HostInfoProvider`` (see Default Component below). No synthetic/heuristic Components are created from namespaces. -- **Apps** - each ROS 2 node becomes an App with ``source: "heuristic"``. +- **Apps** - each ROS 2 node the graph still attributes an endpoint to becomes + an App with ``source: "heuristic"`` (see `What Makes a Node an App`_). - **Functions** - namespace grouping creates Function entities (see below). Default Component @@ -673,6 +674,73 @@ staleness behavior: plugins.parameter_beacon.beacon_ttl_sec: 15.0 plugins.parameter_beacon.beacon_expiry_sec: 300.0 +What Makes a Node an App +------------------------ + +Runtime discovery lists the node names on the graph and then asks the graph +about each name in turn. A name becomes an App only when that second question +comes back with at least one endpoint: a service, a publisher or a +subscription. A name the graph attributes nothing to is not turned into an App, +and where the same App is also declared in a manifest it is linked as +``x-medkit.is_online: false`` instead. + +The rule exists because a name on the graph is not by itself evidence that the +node is there. Node names and endpoints live in different maps inside the RMW +graph cache, filled and emptied by different code paths, and the two can +disagree: a cache can go on naming a node whose endpoints it has already +removed, and it does not correct itself on a timer, because the removal event +for a participant is generated once. A gateway that trusted the name alone +would keep serving that App for as long as the process runs. Asking about the +endpoints costs nothing extra in the normal case - discovery already reads each +node's services to build its operations - and it answers the question the name +cannot. + +The same answer covers the ordinary race. Anything may happen between listing +the names and asking about one of them, including the node exiting; rcl then +reports the name as non-existent and raises. That is read the same way: the +node is not part of this pass, the pass finishes normally, and the next pass +decides again from a fresh read. + +The boundary, stated as a limit rather than as a promise: a node that +advertises no service, no publisher and no subscription at all is not visible as +an App. + +What an rclcpp node puts on the graph, and what a ``NodeOptions`` flag can take +away: + +.. list-table:: + :header-rows: 1 + :widths: 40 25 35 + + * - Entity + - Distro + - Switched off by + * - the six parameter services + - all + - ``start_parameter_services(false)`` + * - ``/parameter_events`` publisher + - all + - ``start_parameter_event_publisher(false)`` + * - ``/rosout`` publisher + - all + - ``enable_rosout(false)`` + * - ``/parameter_events`` subscription (the node's time source watches + ``use_sim_time``) + - all + - nothing - no ``NodeOptions`` flag reaches it + * - ``~/get_type_description`` service + - Jazzy and newer + - the read-only ``start_type_description_service`` parameter + +So a node that switches off everything ``NodeOptions`` offers is still visible: +its time-source subscription alone keeps it an App on every supported distro. +Reaching the boundary takes a node built below rclcpp - an rcl-level node with +no endpoints of any kind - or an rclcpp node whose time source has been taken +away. The fixture ``demo_silent_node`` in ``ros2_medkit_integration_tests`` +carries exactly one endpoint of its own, the ``/rosout`` publisher, and its test +asserts that the graph still attributes it to the node and that the gateway +still lists it. + See Also -------- diff --git a/docs/tutorials/heuristic-apps.rst b/docs/tutorials/heuristic-apps.rst index 8d4b74638..a53a5fd04 100644 --- a/docs/tutorials/heuristic-apps.rst +++ b/docs/tutorials/heuristic-apps.rst @@ -113,7 +113,8 @@ Entity Model In runtime mode, the gateway maps the ROS 2 graph as follows: -- **Apps** - each ROS 2 node becomes an App (``source: "heuristic"``) +- **Apps** - each ROS 2 node the graph still attributes an endpoint to becomes + an App (``source: "heuristic"``); see :doc:`/config/discovery-options` - **Functions** - namespace grouping creates Function entities - **Components** - a single host-level Component from ``HostInfoProvider`` - **Areas** - not created (Areas come from manifest only) diff --git a/src/ros2_medkit_discovery_plugins/ros2_medkit_param_beacon/src/param_beacon_plugin.cpp b/src/ros2_medkit_discovery_plugins/ros2_medkit_param_beacon/src/param_beacon_plugin.cpp index a00dc5ff5..80eaf4156 100644 --- a/src/ros2_medkit_discovery_plugins/ros2_medkit_param_beacon/src/param_beacon_plugin.cpp +++ b/src/ros2_medkit_discovery_plugins/ros2_medkit_param_beacon/src/param_beacon_plugin.cpp @@ -121,6 +121,16 @@ void ParameterBeaconPlugin::set_context(PluginContext & context) { options.start_parameter_event_publisher(false); options.use_global_arguments(false); param_node_ = std::make_shared("_param_beacon_node", options); + // Registered with the context's GraphListener here, while the context is + // known valid. The beacon's parameter sweeps wait for services, and a + // shutdown landing on this node's first such wait would leave it marked as + // registered while absent from the listener's list: + // NodeGraph::get_graph_event() spends should_add_to_graph_listener_ before + // add_node() throws GraphListenerShutdownError, and ~NodeGraph then throws + // NodeNotFoundError out of a noexcept destructor. The window is narrowed, not + // closed: a shutdown between the make_shared above and this line spends the + // flag the same way, and rclcpp offers no way to un-spend it. + (void)param_node_->get_graph_event(); // Set default client factory if not injected (tests inject mock factory) if (!client_factory_) { diff --git a/src/ros2_medkit_gateway/CMakeLists.txt b/src/ros2_medkit_gateway/CMakeLists.txt index 23b20ac4d..9c57aca33 100644 --- a/src/ros2_medkit_gateway/CMakeLists.txt +++ b/src/ros2_medkit_gateway/CMakeLists.txt @@ -882,6 +882,11 @@ if(BUILD_TESTING) target_link_libraries(test_ros2_lifecycle_state_reader gateway_ros2) medkit_target_dependencies(test_ros2_lifecycle_state_reader rclcpp lifecycle_msgs) + # Helper nodes that outlive rclcpp::shutdown() (each test cycles the context) + medkit_add_gtest(test_graph_listener_shutdown_safety test/test_graph_listener_shutdown_safety.cpp) + target_link_libraries(test_graph_listener_shutdown_safety gateway_ros2) + medkit_target_dependencies(test_graph_listener_shutdown_safety rclcpp lifecycle_msgs) + # Add operation handler tests medkit_add_gtest(test_operation_handlers test/test_operation_handlers.cpp) target_link_libraries(test_operation_handlers gateway_ros2) diff --git a/src/ros2_medkit_gateway/README.md b/src/ros2_medkit_gateway/README.md index 5c0f55990..fc3725fd4 100644 --- a/src/ros2_medkit_gateway/README.md +++ b/src/ros2_medkit_gateway/README.md @@ -1790,7 +1790,9 @@ In addition to standard ROS 2 node discovery, the gateway supports **topic-based In runtime discovery mode, the gateway maps the ROS 2 graph to the SOVD entity model: - **Component**: A single host-derived Component is created from `HostInfoProvider` (hostname, OS, architecture). All Apps belong to this Component. -- **App**: Each discovered ROS 2 node becomes an App entity. +- **App**: Each discovered ROS 2 node becomes an App entity, as long as the graph + still attributes at least one service, publisher or subscription to it - a name + with no endpoints left is a node that has gone, not an App. - **Function**: The first namespace segment creates a Function entity that groups all Apps under that namespace (e.g., `/powertrain/engine/temp_sensor` and `/powertrain/engine/rpm_sensor` both belong to Function `powertrain`). - **Area**: Areas are only created from manifest definitions. They are never auto-generated in runtime mode. Use hybrid or manifest-only mode to organize entities into Areas. diff --git a/src/ros2_medkit_gateway/design/lifecycle.rst b/src/ros2_medkit_gateway/design/lifecycle.rst index 6a8462f5a..7509c05ec 100644 --- a/src/ros2_medkit_gateway/design/lifecycle.rst +++ b/src/ros2_medkit_gateway/design/lifecycle.rst @@ -94,10 +94,22 @@ plain node with no lifecycle services, the status falls back to ``App::is_online the ROS 2 graph), which is the best signal available for an unmanaged node. A node that is not online short-circuits to ``"notReady"`` without a GetState read (an offline node cannot be ``active``), which also avoids a blocking read against a crashed managed node whose services still -linger in the cache. The GetState read runs on a private node and executor (spun inline), so it -never blocks or races the gateway executor; it is, however, serialized by an internal mutex, so a -reachable-but-slow managed node holds that mutex across its spin and delays other concurrent -``/status`` reads for up to the (short) read timeout. +linger in the cache. The GetState read runs on a private node, and on an executor created for that one call and spun +inline, so it never blocks or races the gateway executor. That private node is named after the +gateway, so there is exactly one reader per gateway: +``GatewayNode::get_lifecycle_state_reader()`` creates it on first use and both the ``/status`` +handler and any plugin that reads lifecycle state (through +``RosPluginContext::lifecycle_state_reader()``) share it. + +Sharing one object between an HTTP handler and a plugin tick is only safe because the reader's +mutex covers just the two things rclcpp does not make thread-safe - creating and destroying the +call's callback group and client on the shared node. The service wait, the request and the spin +run outside it, so a read of one node never waits for a read of another. That matters because the +graph watchdog's lifecycle watcher seeds nodes that may never answer: with the mutex spanning the +spin, a ``/status`` read of a healthy node was measured queueing behind one such seed for the full +read timeout, against single-digit milliseconds once the mutex was narrowed +(``test_lifecycle_reader_contention_e2e``). Destruction is not ordered by that mutex either - a +call spends most of its life outside it - but by an in-flight count the destructor waits on. **Component status:** the synthetic host component (the one carrying ``host_metadata``, populated by ``HostInfoProvider``) is ``"ready"`` while the gateway is serving the request - diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp index a46a11bf5..aa64f3aa3 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -50,6 +51,7 @@ #include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" #include "ros2_medkit_gateway/core/resource_change_notifier.hpp" #include "ros2_medkit_gateway/core/resource_sampler.hpp" +#include "ros2_medkit_gateway/core/status/lifecycle_state_reader.hpp" #include "ros2_medkit_gateway/core/subscription_transport.hpp" #include "ros2_medkit_gateway/core/trigger_store.hpp" #include "ros2_medkit_gateway/discovery/discovery_manager.hpp" @@ -196,6 +198,18 @@ class GatewayNode : public rclcpp::Node { */ EntityFreezeFrameCapture * get_entity_freeze_frame_capture() const; + /** + * @brief The gateway's lifecycle-state reader, created on first use. + * + * One instance per gateway. The reader owns a private ROS node named after + * this one, so a second instance would put a second node of that exact name + * on the graph: DDS warns about the collision, and every graph query that + * lists nodes then returns the name twice. + * + * @return Shared pointer; never null. + */ + std::shared_ptr get_lifecycle_state_reader(); + /** * @brief Route the trigger topic subscriber through the shared subscription * executor (issue #548). Its per-trigger subscriptions are then @@ -418,6 +432,11 @@ class GatewayNode : public rclcpp::Node { std::unique_ptr trigger_fault_subscriber_; // Zero-config freeze-frames for plugin-backed entities (nullptr when disabled) std::unique_ptr entity_freeze_frame_capture_; + // Shared by the /status handler and by any plugin that reads lifecycle state, + // guarded because the REST server and the plugin manager reach it from + // different threads during start-up. + std::shared_ptr lifecycle_state_reader_; + std::mutex lifecycle_state_reader_mutex_; // Config-less threshold-rule engine + its dedicated evaluation loop (issue // #235). The thread is joined in ~GatewayNode BEFORE plugin/fault shutdown. std::unique_ptr fault_trigger_engine_; diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/plugins/ros_plugin_context.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/plugins/ros_plugin_context.hpp index 783dc3103..0c2b471f3 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/plugins/ros_plugin_context.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/plugins/ros_plugin_context.hpp @@ -17,6 +17,7 @@ #include #include "ros2_medkit_gateway/core/plugins/plugin_context.hpp" +#include "ros2_medkit_gateway/core/status/lifecycle_state_reader.hpp" namespace rclcpp { class Node; @@ -43,6 +44,17 @@ class RosPluginContext : public PluginContext { public: /// Get the ROS 2 node pointer for subscriptions, service clients, etc. virtual rclcpp::Node * node() const = 0; + + /// The gateway's own lifecycle-state reader, for plugins that read managed + /// nodes' states. Shared on purpose: the reader owns a private ROS node named + /// after the gateway, so a plugin that built its own would put a second node + /// of that exact name on the graph. + /// + /// Returns nullptr for contexts that are not backed by a gateway (test + /// doubles); a plugin that gets nullptr owns the fallback. + virtual std::shared_ptr lifecycle_state_reader() const { + return nullptr; + } }; /** diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/providers/ros2_runtime_introspection.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/providers/ros2_runtime_introspection.hpp index 9a2a4eaff..4679a7254 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/providers/ros2_runtime_introspection.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/providers/ros2_runtime_introspection.hpp @@ -96,6 +96,24 @@ class Ros2RuntimeIntrospection : public IntrospectionProvider { /// call from hot paths. std::vector discover_apps(); + /// Services the graph attributes to the node `name` in namespace `ns`, + /// including the internal parameter services, or nullopt when the graph + /// attributes no service, publisher or subscription to it at all. + /// + /// Nullopt is the "this name is not a node any more" answer, and it covers + /// both shapes a departure takes. rcl answers the per-node query for a name + /// it no longer knows with RCL_RET_NODE_NAME_NON_EXISTENT, which rclcpp + /// raises; and an rmw graph cache can keep a node name after the node's + /// endpoints have already been removed from the same cache, in which case + /// every per-node query comes back empty instead. Both mean the node is not + /// there, so both read the same way here. + /// + /// The publisher and subscription queries only run when the node reports no + /// services, so a node with the default parameter services costs exactly one + /// query, as before. + std::optional>> services_of_present_node(const std::string & name, + const std::string & ns) const; + /// Group nodes by namespace into Function entities (no graph query). std::vector discover_functions(const std::vector & apps); diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp index ac561636d..36c6f45b1 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp @@ -15,6 +15,7 @@ #pragma once #include +#include #include #include #include @@ -27,17 +28,24 @@ namespace ros2_medkit_gateway { /// LifecycleStateReader backed by lifecycle_msgs/srv/GetState. The GetState client -/// runs on a private node driven by a private SingleThreadedExecutor that is spun -/// inline on the calling thread (no background spin), so it never races the host -/// gateway node's MultiThreadedExecutor (the private-node/private-executor idea is -/// borrowed from ros2_fault_service_transport.cpp; unlike that transport, the target -/// service path varies per app, so the client is created per call rather than once -/// in the constructor). create_client, async_send_request, the inline spin, and the -/// client teardown are serialized by an internal mutex; wait_for_service runs outside -/// it (backed by an independent graph listener) so an unreachable node does not hold -/// the mutex. A reachable-but-slow node still holds the mutex across its spin for up to -/// the timeout and serializes other concurrent /status reads, so the default timeout is -/// kept short. +/// runs on a private node, never on the host gateway node, so it cannot race that +/// node's MultiThreadedExecutor (the private-node idea is borrowed from +/// ros2_fault_service_transport.cpp; unlike that transport, the target service path +/// varies per app, so the client is created per call rather than once in the +/// constructor). +/// +/// One instance serves every caller in the process - the /status handler and any +/// plugin that reads lifecycle state - so a slow or unanswering target must not be +/// able to delay a caller asking about a different node. Each call therefore gets its +/// own callback group and its own executor, spun inline on the calling thread, and the +/// internal mutex covers only what rclcpp does not make thread-safe: creating and +/// destroying that group and client on the shared node. The service wait, the request +/// and the spin all run outside it, so concurrent reads overlap. +/// +/// The mutex is not what makes destruction safe, since a call spends most of its life +/// outside it. `in_flight_` is: the destructor refuses new calls and waits for the +/// ones already running, so the private node outlives every executor that references +/// it. class Ros2LifecycleStateReader : public LifecycleStateReader { public: explicit Ros2LifecycleStateReader(rclcpp::Node * host, @@ -52,9 +60,15 @@ class Ros2LifecycleStateReader : public LifecycleStateReader { private: std::shared_ptr client_node_; - std::shared_ptr executor_; std::chrono::duration timeout_; + /// Guards client_node_'s callback-group and client registries, which rclcpp + /// does not serialize, plus the two fields below. std::mutex mutex_; + std::condition_variable idle_cv_; + /// Calls that have created their client and not yet destroyed it. + int in_flight_{0}; + /// Set by the destructor; a call that sees it returns without touching the node. + bool stopping_{false}; }; } // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2_common/callback_groups.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2_common/callback_groups.hpp index 12fbdac59..9df8d1cfa 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2_common/callback_groups.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2_common/callback_groups.hpp @@ -72,4 +72,22 @@ struct GatewayCallbackGroups { */ GatewayCallbackGroups create_gateway_callback_groups(rclcpp::Node & node); +/** + * @brief A MutuallyExclusive group on @p node that no executor collects along + * with the node. + * + * For an entity created, used and destroyed inside a single call on a single + * thread. `automatically_add_to_executor_with_node = false` means the group is + * reachable only from the executor that call builds for it, so no other thread + * ever holds a reference to the entity and the entity can be destroyed on the + * calling thread rather than on an executor thread - the ordering rule the + * gateway relies on everywhere it creates ROS entities while running. + * + * Unlike `create_gateway_callback_groups`, this is called while the node is + * live, so the caller owns the serialisation: every call must be made under the + * same lock as every other node-mutating call on @p node. Group creation is one + * of the rcl hash-map mutations the issue-#375 gate exists for. + */ +rclcpp::CallbackGroup::SharedPtr create_isolated_callback_group(rclcpp::Node & node); + } // namespace ros2_medkit_gateway::ros2_common diff --git a/src/ros2_medkit_gateway/src/gateway_node.cpp b/src/ros2_medkit_gateway/src/gateway_node.cpp index a8f32986e..d0993369f 100644 --- a/src/ros2_medkit_gateway/src/gateway_node.cpp +++ b/src/ros2_medkit_gateway/src/gateway_node.cpp @@ -38,6 +38,7 @@ #include "ros2_medkit_gateway/core/thread_pool_config.hpp" #include "ros2_medkit_gateway/param_utils.hpp" #include "ros2_medkit_gateway/plugins/ros_plugin_context.hpp" +#include "ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp" #include "ros2_medkit_gateway/core/http/handlers/sse_transport_provider.hpp" #include "ros2_medkit_gateway/core/sqlite_trigger_store.hpp" @@ -1915,6 +1916,14 @@ EntityFreezeFrameCapture * GatewayNode::get_entity_freeze_frame_capture() const return entity_freeze_frame_capture_.get(); } +std::shared_ptr GatewayNode::get_lifecycle_state_reader() { + std::lock_guard lock(lifecycle_state_reader_mutex_); + if (!lifecycle_state_reader_) { + lifecycle_state_reader_ = std::make_shared(this); + } + return lifecycle_state_reader_; +} + void GatewayNode::set_trigger_subscription_executor(ros2_common::Ros2SubscriptionExecutor & exec) { if (trigger_topic_subscriber_) { trigger_topic_subscriber_->set_subscription_executor(&exec); diff --git a/src/ros2_medkit_gateway/src/http/rest_server.cpp b/src/ros2_medkit_gateway/src/http/rest_server.cpp index 5b070aece..113f2b219 100644 --- a/src/ros2_medkit_gateway/src/http/rest_server.cpp +++ b/src/ros2_medkit_gateway/src/http/rest_server.cpp @@ -32,7 +32,6 @@ #include "ros2_medkit_gateway/dto/sse_frames.hpp" #include "ros2_medkit_gateway/gateway_node.hpp" #include "ros2_medkit_gateway/http/detail/status_recorder.hpp" -#include "ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp" #include "../openapi/route_registry.hpp" #include "../openapi/schema_builder.hpp" @@ -183,8 +182,8 @@ RESTServer::RESTServer(GatewayNode * node, const std::string & host, int port, c std::make_unique(*handler_ctx_, route_registry_.get(), sse_client_tracker_); discovery_handlers_ = std::make_unique(*handler_ctx_); data_handlers_ = std::make_unique(*handler_ctx_); - lifecycle_handlers_ = std::make_unique( - *handler_ctx_, node_->get_plugin_manager(), std::make_shared(node_)); + lifecycle_handlers_ = std::make_unique(*handler_ctx_, node_->get_plugin_manager(), + node_->get_lifecycle_state_reader()); operation_handlers_ = std::make_unique(*handler_ctx_); config_handlers_ = std::make_unique(*handler_ctx_); fault_handlers_ = std::make_unique(*handler_ctx_); diff --git a/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp b/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp index 47bc64937..013ff101f 100644 --- a/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp +++ b/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp @@ -69,6 +69,10 @@ class GatewayPluginContext : public RosPluginContext { return node_; } + std::shared_ptr lifecycle_state_reader() const override { + return node_->get_lifecycle_state_reader(); + } + std::optional get_entity(const std::string & id) const override { const auto & cache = node_->get_thread_safe_cache(); diff --git a/src/ros2_medkit_gateway/src/ros2/providers/ros2_runtime_introspection.cpp b/src/ros2_medkit_gateway/src/ros2/providers/ros2_runtime_introspection.cpp index 66381e166..fb6bf0d73 100644 --- a/src/ros2_medkit_gateway/src/ros2/providers/ros2_runtime_introspection.cpp +++ b/src/ros2_medkit_gateway/src/ros2/providers/ros2_runtime_introspection.cpp @@ -124,6 +124,13 @@ std::vector Ros2RuntimeIntrospection::discover_apps() { } seen_fqns.insert(fqn); + auto node_services_opt = services_of_present_node(name, ns); + if (!node_services_opt) { + RCLCPP_DEBUG(node_->get_logger(), "Node '%s' advertises no endpoint; not an app in this pass", fqn.c_str()); + continue; + } + const auto & node_services = *node_services_opt; + App app; if (name_namespaces[name].size() > 1 && ns != "/") { std::string ns_prefix = ns.substr(1); // Remove leading '/' @@ -141,45 +148,39 @@ std::vector Ros2RuntimeIntrospection::discover_apps() { app.is_online = true; app.bound_fqn = fqn; - try { - auto node_services = node_->get_service_names_and_types_by_node(name, ns); - for (const auto & [service_path, types] : node_services) { - if (is_internal_service(service_path)) { - continue; - } - auto it = service_info_map.find(service_path); - if (it != service_info_map.end()) { - app.services.push_back(it->second); - } else { - ServiceInfo info; - info.full_path = service_path; - info.name = extract_name_from_path(service_path); - info.type = types.empty() ? "" : types[0]; - app.services.push_back(info); - } + for (const auto & [service_path, types] : node_services) { + if (is_internal_service(service_path)) { + continue; + } + auto it = service_info_map.find(service_path); + if (it != service_info_map.end()) { + app.services.push_back(it->second); + } else { + ServiceInfo info; + info.full_path = service_path; + info.name = extract_name_from_path(service_path); + info.type = types.empty() ? "" : types[0]; + app.services.push_back(info); } + } - // Detect actions by checking for /_action/send_goal services - for (const auto & [service_path, types] : node_services) { - const std::string action_suffix = "/_action/send_goal"; - if (service_path.length() > action_suffix.length() && - service_path.compare(service_path.length() - action_suffix.length(), action_suffix.length(), - action_suffix) == 0) { - std::string action_path = service_path.substr(0, service_path.length() - action_suffix.length()); - auto it = action_info_map.find(action_path); - if (it != action_info_map.end()) { - app.actions.push_back(it->second); - } else { - ActionInfo info; - info.full_path = action_path; - info.name = extract_name_from_path(action_path); - app.actions.push_back(info); - } + // Detect actions by checking for /_action/send_goal services + for (const auto & [service_path, types] : node_services) { + const std::string action_suffix = "/_action/send_goal"; + if (service_path.length() > action_suffix.length() && + service_path.compare(service_path.length() - action_suffix.length(), action_suffix.length(), action_suffix) == + 0) { + std::string action_path = service_path.substr(0, service_path.length() - action_suffix.length()); + auto it = action_info_map.find(action_path); + if (it != action_info_map.end()) { + app.actions.push_back(it->second); + } else { + ActionInfo info; + info.full_path = action_path; + info.name = extract_name_from_path(action_path); + app.actions.push_back(info); } } - } catch (const std::exception & e) { - RCLCPP_DEBUG(node_->get_logger(), "Could not get services for node '%s' in namespace '%s': %s", name.c_str(), - ns.c_str(), e.what()); } if (topic_data_provider_) { @@ -196,6 +197,33 @@ std::vector Ros2RuntimeIntrospection::discover_apps() { return apps; } +std::optional>> +Ros2RuntimeIntrospection::services_of_present_node(const std::string & name, const std::string & ns) const { + std::map> services; + try { + services = node_->get_service_names_and_types_by_node(name, ns); + if (!services.empty()) { + return services; + } + // Every rclcpp and rclpy node carries the six parameter services unless + // they were turned off explicitly, so an empty service map is already the + // unusual case. Confirm it against the node's other endpoints before + // treating the name as gone. + auto node_graph = node_->get_node_graph_interface(); + if (!node_graph->get_publisher_names_and_types_by_node(name, ns).empty()) { + return services; + } + if (!node_graph->get_subscriber_names_and_types_by_node(name, ns).empty()) { + return services; + } + } catch (const std::exception & e) { + RCLCPP_DEBUG(node_->get_logger(), "Per-node graph query for '%s' in namespace '%s' failed: %s", name.c_str(), + ns.c_str(), e.what()); + return std::nullopt; + } + return std::nullopt; +} + std::vector Ros2RuntimeIntrospection::discover_functions() { if (!config_.create_functions_from_namespaces) { return {}; diff --git a/src/ros2_medkit_gateway/src/ros2/status/ros2_lifecycle_state_reader.cpp b/src/ros2_medkit_gateway/src/ros2/status/ros2_lifecycle_state_reader.cpp index 92baba50d..509768bbb 100644 --- a/src/ros2_medkit_gateway/src/ros2/status/ros2_lifecycle_state_reader.cpp +++ b/src/ros2_medkit_gateway/src/ros2/status/ros2_lifecycle_state_reader.cpp @@ -14,29 +14,59 @@ #include "ros2_medkit_gateway/ros2/status/ros2_lifecycle_state_reader.hpp" +#include "ros2_medkit_gateway/ros2_common/callback_groups.hpp" + #include #include #include +#include namespace ros2_medkit_gateway { +namespace { + +/// create_client() with an explicit callback group, spelled the way each +/// supported distro wants it. Jazzy is rclcpp 28 and takes rclcpp::QoS; older +/// distros only offer the rmw_qos_profile_t form, where it is not deprecated. +rclcpp::Client::SharedPtr +create_get_state_client(rclcpp::Node * node, const std::string & service_name, + const rclcpp::CallbackGroup::SharedPtr & group) { +#if defined(RCLCPP_VERSION_MAJOR) && RCLCPP_VERSION_MAJOR >= 28 + return node->create_client(service_name, rclcpp::ServicesQoS(), group); +#else + return node->create_client(service_name, rmw_qos_profile_services_default, group); +#endif +} + +} // namespace + Ros2LifecycleStateReader::Ros2LifecycleStateReader(rclcpp::Node * host, std::chrono::duration timeout) : timeout_(timeout) { client_node_ = std::make_shared(std::string(host->get_name()) + "_lifecycle_state_reader"); - executor_ = std::make_shared(); - executor_->add_node(client_node_); + // Registered with the context's GraphListener here, while the context is + // known valid. Left to the first graph use, NodeGraph::get_graph_event() + // would spend should_add_to_graph_listener_ before add_node() throws + // GraphListenerShutdownError on a stopped listener; the node is then marked + // as registered while absent from the listener's list, and ~NodeGraph turns + // that into a NodeNotFoundError thrown out of a noexcept destructor. The + // window is narrowed, not closed: a shutdown between the make_shared above + // and this line spends the flag the same way, and rclcpp offers no way to + // un-spend it. + (void)client_node_->get_graph_event(); } Ros2LifecycleStateReader::~Ros2LifecycleStateReader() { - // Tear down under the mutex so that an in-flight get_state() (which holds the - // mutex across async_send_request + spin) has finished first: remove_node - // during an active spin is undefined behavior. Mirrors the executor-mutex - // teardown in ros2_fault_service_transport.cpp. - std::lock_guard lock(mutex_); - if (executor_ && client_node_) { - executor_->remove_node(client_node_); - } + // A call spends most of its life outside mutex_, so holding the mutex here + // would prove nothing about the node being idle. Refuse new calls, then wait + // for the ones already running: each of them owns an executor that holds a + // reference to client_node_, and that executor is destroyed before the call + // decrements in_flight_. + std::unique_lock lock(mutex_); + stopping_ = true; + idle_cv_.wait(lock, [this] { + return in_flight_ == 0; + }); } std::optional Ros2LifecycleStateReader::get_state(const std::string & get_state_service_path) { @@ -50,41 +80,59 @@ std::optional Ros2LifecycleStateReader::get_state(const std::string } const auto clamped = std::chrono::duration(std::max(timeout_.count(), 0.0)); - // create_client (and ~Client at the tail) mutate client_node_'s registry, so - // they are serialized by mutex_. wait_for_service is backed by an independent - // graph listener and does not touch the executor, so it runs OUTSIDE the - // mutex: otherwise a slow or unreachable lifecycle node would hold the mutex - // for the whole timeout and head-of-line-block every other concurrent - // /status read (ros2_fault_service_transport.cpp keeps wait_for_service - // outside its executor mutex for the same reason). + // Making the call's callback group and its client mutates registries on the + // shared node that rclcpp does not serialize, and so does destroying them. + // Those four moments are all mutex_ covers. Everything that can block - the + // service wait, the request, the spin - happens outside it, because one + // instance serves every caller in the process and an unanswering target must + // not hold up a caller asking about a different node. + rclcpp::CallbackGroup::SharedPtr group; rclcpp::Client::SharedPtr client; { std::lock_guard lock(mutex_); + if (stopping_) { + return std::nullopt; + } try { - client = client_node_->create_client(get_state_service_path); + // An isolated group keeps this client out of any executor the private + // node is added to, so the only thread that ever holds a reference to it + // is the one running this call - which is what lets the client be + // destroyed here rather than on an executor thread. + group = ros2_common::create_isolated_callback_group(*client_node_); + client = create_get_state_client(client_node_.get(), get_state_service_path, group); } catch (const std::exception & e) { RCLCPP_WARN(client_node_->get_logger(), "GetState client creation failed for '%s': %s", get_state_service_path.c_str(), e.what()); return std::nullopt; } + ++in_flight_; } std::optional label; - if (client->wait_for_service(clamped)) { - auto request = std::make_shared(); - std::lock_guard lock(mutex_); - auto future = client->async_send_request(request); - if (executor_->spin_until_future_complete(future, clamped) == rclcpp::FutureReturnCode::SUCCESS) { - label = future.get()->current_state.label; - } else { - client->remove_pending_request(future.request_id); + { + // This call's own executor, spun inline on this thread and destroyed on it. + rclcpp::executors::SingleThreadedExecutor executor; + executor.add_callback_group(group, client_node_->get_node_base_interface()); + if (client->wait_for_service(clamped)) { + auto request = std::make_shared(); + auto future = client->async_send_request(request); + if (executor.spin_until_future_complete(future, clamped) == rclcpp::FutureReturnCode::SUCCESS) { + label = future.get()->current_state.label; + } else { + // Drop the abandoned slot from the client's pending-request map. + client->remove_pending_request(future.request_id); + } } + executor.remove_callback_group(group); } - // Destroy the client under the mutex: like create_client, ~Client mutates - // client_node_'s registry and would race a concurrent create_client. - std::lock_guard lock(mutex_); - client.reset(); + { + std::lock_guard lock(mutex_); + client.reset(); + group.reset(); + --in_flight_; + } + idle_cv_.notify_all(); return label; } diff --git a/src/ros2_medkit_gateway/src/ros2/transports/ros2_fault_service_transport.cpp b/src/ros2_medkit_gateway/src/ros2/transports/ros2_fault_service_transport.cpp index e284254a2..87b6eb6ee 100644 --- a/src/ros2_medkit_gateway/src/ros2/transports/ros2_fault_service_transport.cpp +++ b/src/ros2_medkit_gateway/src/ros2/transports/ros2_fault_service_transport.cpp @@ -102,6 +102,22 @@ Ros2FaultServiceTransport::Ros2FaultServiceTransport(rclcpp::Node * node) : node // race against the calling thread destroying the response shared_ptr - both // happen inline on the caller's thread inside spin_until_future_complete(). client_node_ = std::make_shared(std::string(node_->get_name()) + "_fault_clients"); + // Register client_node_ with the context's GraphListener here, where the + // context is known valid, rather than leaving it to the first + // wait_for_service. NodeGraph::get_graph_event() spends + // should_add_to_graph_listener_ BEFORE calling add_node(), and add_node() + // throws GraphListenerShutdownError once rclcpp::shutdown() has stopped the + // listener. The flag is then spent on a node that was never listed, so + // ~NodeGraph takes its remove_node() branch, the node is absent from + // node_graph_interfaces_, and NodeNotFoundError escapes a noexcept destructor + // -> std::terminate, exit -6. This narrows the window rather than closing it: + // a shutdown landing between the make_shared above and this line still spends + // the flag inside a throwing constructor, and rclcpp offers no way to un-spend + // it. What it removes is the part that is ordinary - a shutdown arriving + // during a fault service wait, which lasts as long as the wait does. The + // returned event is not needed: nothing here waits on graph changes, and + // NodeGraph holds it weakly. + (void)client_node_->get_graph_event(); executor_ = std::make_shared(); executor_->add_node(client_node_); diff --git a/src/ros2_medkit_gateway/src/ros2_common/callback_groups.cpp b/src/ros2_medkit_gateway/src/ros2_common/callback_groups.cpp index c2f6a12e3..997fbec44 100644 --- a/src/ros2_medkit_gateway/src/ros2_common/callback_groups.cpp +++ b/src/ros2_medkit_gateway/src/ros2_common/callback_groups.cpp @@ -23,4 +23,9 @@ GatewayCallbackGroups create_gateway_callback_groups(rclcpp::Node & node) { return groups; } +rclcpp::CallbackGroup::SharedPtr create_isolated_callback_group(rclcpp::Node & node) { + return node.create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive, + /*automatically_add_to_executor_with_node=*/false); +} + } // namespace ros2_medkit_gateway::ros2_common diff --git a/src/ros2_medkit_gateway/test/test_graph_listener_shutdown_safety.cpp b/src/ros2_medkit_gateway/test/test_graph_listener_shutdown_safety.cpp new file mode 100644 index 000000000..eb69ef2d5 --- /dev/null +++ b/src/ros2_medkit_gateway/test/test_graph_listener_shutdown_safety.cpp @@ -0,0 +1,76 @@ +// Copyright 2026 bburda +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include +#include + +#include +#include + +#include "ros2_medkit_gateway/ros2/transports/ros2_fault_service_transport.hpp" + +namespace { + +using ros2_medkit_gateway::ros2::Ros2FaultServiceTransport; + +// The gateway's helper classes each own a private rclcpp::Node whose first use +// of the ROS graph can land after rclcpp::shutdown() has stopped the context's +// GraphListener - a shutdown signal while a service wait is in flight is the +// ordinary case. The test runs a complete init/shutdown cycle of the default +// context and exercises the helper across that boundary. +// +// Only Ros2FaultServiceTransport is driven here, and it is the only one of the +// three helper nodes this gives a falsifying test. Its clients are built in its +// constructor, so a post-shutdown wait_for_service reaches +// NodeGraph::get_graph_event() and the registration that follows it. +// Ros2LifecycleStateReader and ParameterBeaconPlugin build their clients per +// call, so after shutdown rcl_client_init fails first and the wait is never +// reached: for those two the eager registration is correct by construction and +// no test at this tier can tell the two versions apart. +class GraphListenerShutdownSafetyTest : public ::testing::Test { + protected: + void SetUp() override { + rclcpp::init(0, nullptr); + } + + void TearDown() override { + if (rclcpp::ok()) { + rclcpp::shutdown(); + } + } +}; + +TEST_F(GraphListenerShutdownSafetyTest, FaultTransportSurvivesAServiceWaitAfterShutdown) { + auto host = std::make_shared("fault_transport_host"); + // The gateway node takes a graph event while it is being built, which is what + // brings the context's GraphListener into existence and starts its thread. + // rclcpp::shutdown() then stops that listener rather than leaving the helper + // node to create a fresh one. + auto host_event = host->get_graph_event(); + auto transport = std::make_unique(host.get()); + + rclcpp::shutdown(); + + // No fault manager can be reached on a shut-down context, so the wait reports + // the services as unavailable rather than raising. + EXPECT_NO_THROW({ EXPECT_FALSE(transport->wait_for_services(std::chrono::duration(0.2))); }); + + // The private node is registered with the graph listener, so ~NodeGraph finds + // it and removes it. An unregistered node aborts the process here instead: + // NodeNotFoundError escapes a noexcept destructor. + transport.reset(); + host.reset(); +} + +} // namespace diff --git a/src/ros2_medkit_gateway/test/test_runtime_discovery.cpp b/src/ros2_medkit_gateway/test/test_runtime_discovery.cpp index 6f758f898..597920654 100644 --- a/src/ros2_medkit_gateway/test/test_runtime_discovery.cpp +++ b/src/ros2_medkit_gateway/test/test_runtime_discovery.cpp @@ -220,3 +220,29 @@ TEST_F(RuntimeDiscoveryMultiNsTest, Introspect_AreasAndComponentsEmpty) { EXPECT_TRUE(result.new_entities.components.empty()) << "Components should always be empty - Components come from HostInfoProvider or manifest"; } + +// ----------------------------------------------------------------------------- +// services_of_present_node() - an App is a node the graph still attributes at +// least one endpoint to. +// ----------------------------------------------------------------------------- + +TEST_F(RuntimeDiscoveryTest, PresentNodeIsReportedWithItsServices) { + auto services = strategy_->services_of_present_node(node_->get_name(), node_->get_namespace()); + ASSERT_TRUE(services.has_value()) << "A live node with default parameter services must read as present"; + EXPECT_FALSE(services->empty()); +} + +TEST_F(RuntimeDiscoveryTest, NodeNameThatIsNotOnTheGraphIsReportedAbsent) { + // rcl answers a per-node query for an unknown node with + // RCL_RET_NODE_NAME_NON_EXISTENT, which rclcpp raises. That is what a node + // leaving the graph between the name listing and the query looks like. + EXPECT_FALSE(strategy_->services_of_present_node("node_that_left_the_graph", "/test_ns").has_value()); +} + +TEST_F(RuntimeDiscoveryTest, DiscoverAppsListsTheLiveNode) { + auto apps = strategy_->discover_apps(); + const bool found = std::any_of(apps.begin(), apps.end(), [](const auto & app) { + return app.bound_fqn.has_value() && *app.bound_fqn == "/test_ns/test_node"; + }); + EXPECT_TRUE(found) << "A node the graph attributes endpoints to stays an App"; +} diff --git a/src/ros2_medkit_integration_tests/CMakeLists.txt b/src/ros2_medkit_integration_tests/CMakeLists.txt index dab635448..975318e0c 100644 --- a/src/ros2_medkit_integration_tests/CMakeLists.txt +++ b/src/ros2_medkit_integration_tests/CMakeLists.txt @@ -118,6 +118,10 @@ add_executable(demo_unresponsive_param_node demo_nodes/unresponsive_param_node.c target_include_directories(demo_unresponsive_param_node PRIVATE ${_demo_include_dir}) medkit_target_dependencies(demo_unresponsive_param_node rclcpp rcl_interfaces) +add_executable(demo_silent_node demo_nodes/silent_node.cpp) +target_include_directories(demo_silent_node PRIVATE ${_demo_include_dir}) +medkit_target_dependencies(demo_silent_node rclcpp std_msgs) + install(TARGETS demo_engine_temp_sensor demo_engine_temp_monitor @@ -137,6 +141,7 @@ install(TARGETS managed_lifecycle unreadable_lifecycle demo_unresponsive_param_node + demo_silent_node DESTINATION lib/${PROJECT_NAME} ) @@ -352,7 +357,12 @@ if(BUILD_TESTING) # broken merge spends all of them. The glob default cuts that short and # reports a timeout with no test name, which is the one answer that says # nothing about which rule broke. - test_aggregator_only_configurations 300) + test_aggregator_only_configurations 300 + # 30 churn cycles, each allowed CHURN_NODE_EXIT_TIMEOUT_SEC (30 s) to act on + # its signal, then a departure poll of DEPARTURE_TIMEOUT_SEC (25 s), then the + # quiet-node case. On the glob default a single slow exit eats a quarter of + # the budget and the message naming the cycle that broke never prints. + test_departed_node_discovery 300) # Names actually matched against a discovered test_name in the two loops # below. Checked against _MEDKIT_TEST_TIMEOUT_OVERRIDES itself after both # loops finish - see the FATAL_ERROR check at the end of this block - so a diff --git a/src/ros2_medkit_integration_tests/demo_nodes/silent_node.cpp b/src/ros2_medkit_integration_tests/demo_nodes/silent_node.cpp new file mode 100644 index 000000000..83c0a1080 --- /dev/null +++ b/src/ros2_medkit_integration_tests/demo_nodes/silent_node.cpp @@ -0,0 +1,63 @@ +// Copyright 2026 bburda +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// The control on discovery's rule that a name the graph attributes no endpoint +// to is not an App: that rule may only remove names whose endpoints are gone, +// never a node that is merely quiet. This node turns off everything a node can +// be told to turn off except /rosout - parameter services and the +// parameter-event publisher - and keeps the /rosout publisher, which is the one +// endpoint rclcpp creates the same way on every supported distro. A test can +// therefore name the endpoint it expects to see instead of relying on whichever +// of rclcpp's implicit entities the local distro happens to create. +// +// `advertise:=true` adds a publisher of the node's own, for a case that wants +// an endpoint rclcpp did not create. + +#include +#include + +#include +#include + +#include "ros2_medkit_integration_tests/demo_node_main.hpp" + +class SilentNode : public rclcpp::Node { + public: + explicit SilentNode(const rclcpp::NodeOptions & options) : Node("silent_node", options) { + if (declare_parameter("advertise", false)) { + heartbeat_pub_ = create_publisher("~/heartbeat", 10); + } + } + + ~SilentNode() override { + heartbeat_pub_.reset(); + } + + SilentNode(const SilentNode &) = delete; + SilentNode & operator=(const SilentNode &) = delete; + SilentNode(SilentNode &&) = delete; + SilentNode & operator=(SilentNode &&) = delete; + + private: + rclcpp::Publisher::SharedPtr heartbeat_pub_; +}; + +int main(int argc, char ** argv) { + return ros2_medkit_integration_tests::run_demo_node(argc, argv, [] { + rclcpp::NodeOptions options; + options.start_parameter_services(false); + options.start_parameter_event_publisher(false); + return std::make_shared(options); + }); +} diff --git a/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/launch_helpers.py b/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/launch_helpers.py index f36a329de..23a9ebfd0 100644 --- a/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/launch_helpers.py +++ b/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/launch_helpers.py @@ -76,6 +76,11 @@ # Regression fixture (#531): parameter services are discoverable # (wait_for_service succeeds) but list_parameters never replies. 'unresponsive_param': ('demo_unresponsive_param_node', 'unresponsive_param', ''), + # As little graph surface as a node can be configured to have: parameter + # services, the parameter-event publisher and /rosout are all off. The + # control on discovery's "a name the graph attributes no endpoint to is + # not an App" rule. + 'silent': ('demo_silent_node', 'silent_node', ''), } # Convenience groupings for callers that want subsets of demo nodes. @@ -109,9 +114,16 @@ def create_gateway_node(*, port=DEFAULT_PORT, name='ros2_medkit_gateway', ---------- port : int HTTP server port (default: 8080). - name : str + name : str or None ROS node name. Override when a test launches more than one gateway so their names do not collide (e.g. ``gateway_with_scripts``). + ``None`` omits the name, and with it the ``-r __node:=`` ros-arg + launch_ros would otherwise emit. That remap is a GLOBAL argument, so + rclcpp applies it to every node the gateway process creates, not just + the gateway's own: the private client nodes (``_fault_clients``, + ``_sub``, ``_lifecycle_state_reader``) then all answer to the one name. + Pass ``None`` when a test needs to see those nodes under their own + names; the gateway falls back to its compiled default. extra_params : dict or None Additional ROS parameters merged into the node config. coverage : bool diff --git a/src/ros2_medkit_integration_tests/test/features/test_departed_node_discovery.test.py b/src/ros2_medkit_integration_tests/test/features/test_departed_node_discovery.test.py new file mode 100644 index 000000000..015a15b00 --- /dev/null +++ b/src/ros2_medkit_integration_tests/test/features/test_departed_node_discovery.test.py @@ -0,0 +1,281 @@ +# Copyright 2026 bburda +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Integration tests for a node that leaves the ROS graph while discovery runs. + +A discovery pass lists node names first and then asks the graph about each +name. Anything can happen between those two steps, and what the second step +answers for a name that is no longer a node is either an rcl error or an empty +set of endpoints. Both mean the same thing and both are ordinary, so the +gateway has to keep serving across them. + +Two scenarios: + +1. Churn. A demo node is started and signalled repeatedly at an interval short + enough to land the signal inside the node's own start-up, which is the + interval that puts departures inside discovery passes. Afterwards the + gateway still answers and the node is gone from ``/apps``; its exit code is + checked by the shared shutdown case. + +2. A minimal live node is still an App. ``silent_node`` runs with parameter + services and the parameter-event publisher switched off, keeping only its + ``/rosout`` publisher - the one endpoint rclcpp creates the same way on + every supported distro. The case names that endpoint through an rclpy probe + before asking the gateway, so a pass means "the graph attributes an endpoint + to this node AND the gateway lists it" rather than either half alone. This + is the control on the rule that removes names the graph attributes no + endpoint to: it may only remove names that are gone, never a node that is + merely quiet. +""" + +import os +import signal +import subprocess +import time +import unittest + +from ament_index_python.packages import get_package_prefix +from launch import LaunchDescription +from launch.actions import TimerAction +import launch_testing +import launch_testing.actions +import rclpy + +from ros2_medkit_test_utils.constants import ( + ALLOWED_EXIT_CODES, + DEFAULT_DOMAIN_ID, + get_time_scale, +) +from ros2_medkit_test_utils.gateway_test_case import GatewayTestCase +from ros2_medkit_test_utils.launch_helpers import ( + create_demo_nodes, + create_gateway_node, + DEMO_NODE_REGISTRY, + get_coverage_env, +) + +# The node that is started and signalled over and over. It carries a service, +# so a pass that catches it half-way up or half-way down has per-node queries +# to run against it rather than skipping it on the name alone. +CHURN_NODE_KEY = 'calibration' + +# Cycles, and how long each one lets the node live. A ROS node needs about a +# second to finish announcing itself, so a fraction of that puts the signal +# inside start-up: the departure then lands while the gateway is mid-pass over +# a graph that still names the node. Thirty cycles at that interval is what +# opens the window often enough to be worth running; the count is not a +# statistical claim. +CHURN_CYCLES = 30 +CHURN_NODE_LIFETIME_SEC = 0.1 + +# A churned process is signalled during start-up, so it may have to finish +# coming up before it can act on the signal. +CHURN_NODE_EXIT_TIMEOUT_SEC = 30.0 * get_time_scale() + +# What a departure costs once the process is gone: the DDS participant lease +# plus one gateway refresh. Stated in the gateway's own configuration +# reference; repeated here only as a budget. +DEPARTURE_TIMEOUT_SEC = 25.0 * get_time_scale() + +# Start-up budget for the quiet node, measured from the gateway side. +SILENT_NODE_TIMEOUT_SEC = 30.0 * get_time_scale() + +SILENT_NODE_NAME = 'silent_node' + +# The endpoint the quiet node is guaranteed to own. rclcpp creates the /rosout +# publisher for every node on every supported distro unless enable_rosout is +# turned off, and this fixture leaves it on precisely so the control has an +# endpoint that does not depend on the distro. +SILENT_NODE_ENDPOINT = '/rosout' + +# The probe reads the graph directly, so it is bounded by DDS discovery rather +# than by a gateway refresh. +PROBE_TIMEOUT_SEC = 30.0 * get_time_scale() +PROBE_INTERVAL_SEC = 0.5 + + +def generate_test_description(): + gateway_node = create_gateway_node( + extra_params={ + # Short backstop and short debounce: a pass then starts often + # enough that the churn below runs concurrently with one. + 'refresh_interval_ms': 500, + 'discovery.refresh_debounce_ms': 100, + }, + ) + + delayed = TimerAction( + period=2.0, + actions=create_demo_nodes(nodes=['silent'], lidar_faulty=False), + ) + + return ( + LaunchDescription([ + gateway_node, + delayed, + launch_testing.actions.ReadyToTest(), + ]), + {'gateway_node': gateway_node}, + ) + + +def _resolve_demo_executable(name): + pkg = 'ros2_medkit_integration_tests' + candidate = os.path.join(get_package_prefix(pkg), 'lib', pkg, name) + if not os.path.isfile(candidate): + raise FileNotFoundError(f'demo executable not found: {candidate}') + return candidate + + +class TestDepartedNodeDiscovery(GatewayTestCase): + """A node leaving mid-pass must not cost the gateway anything.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls._churn_procs = [] + + @classmethod + def tearDownClass(cls): + for proc in cls._churn_procs: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + cls._churn_procs = [] + super().tearDownClass() + + @classmethod + def _spawn(cls, key): + executable, ros_name, namespace = DEMO_NODE_REGISTRY[key] + env = os.environ.copy() + env['ROS_DOMAIN_ID'] = str(DEFAULT_DOMAIN_ID) + env.update(get_coverage_env()) + proc = subprocess.Popen( + [ + _resolve_demo_executable(executable), + '--ros-args', + '-r', f'__ns:={namespace}', + '-r', f'__node:={ros_name}', + ], + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + cls._churn_procs.append(proc) + return proc + + def _app_ids(self): + data = self.get_json('/apps') + return [app['id'] for app in data.get('items', [])] + + def _poll_probe_publishers(self, node_name): + """Topics the graph attributes to `node_name`, read with rclpy. + + Asked of the graph rather than of the gateway on purpose: the gateway is + the thing under test, so the anti-vacuity check for this case cannot + come from it. + """ + rclpy.init() + try: + probe = rclpy.create_node('departed_node_discovery_probe') + try: + deadline = time.monotonic() + PROBE_TIMEOUT_SEC + topics = [] + while time.monotonic() < deadline: + for name, namespace in probe.get_node_names_and_namespaces(): + if name != node_name: + continue + topics = [ + topic for topic, _ in + probe.get_publisher_names_and_types_by_node(name, namespace) + ] + if topics: + return topics + time.sleep(PROBE_INTERVAL_SEC) + return topics + finally: + probe.destroy_node() + finally: + rclpy.shutdown() + + def test_01_gateway_serves_through_node_churn(self): + """Repeated departures mid-pass leave the gateway answering.""" + _, ros_name, _ = DEMO_NODE_REGISTRY[CHURN_NODE_KEY] + + for cycle in range(CHURN_CYCLES): + proc = self._spawn(CHURN_NODE_KEY) + time.sleep(CHURN_NODE_LIFETIME_SEC) + proc.send_signal(signal.SIGTERM) + try: + proc.wait(timeout=CHURN_NODE_EXIT_TIMEOUT_SEC) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=10) + self.fail( + f'churn cycle {cycle}: demo node did not act on SIGTERM within ' + f'{CHURN_NODE_EXIT_TIMEOUT_SEC}s' + ) + + health = self.get_json('/health') + self.assertEqual( + health.get('status'), 'healthy', f'gateway unhealthy after churn: {health}') + + departed = self.poll_endpoint_until( + '/apps', + lambda data: data if ros_name not in [ + app['id'] for app in data.get('items', []) + ] else None, + timeout=DEPARTURE_TIMEOUT_SEC, + ) + self.assertIsNotNone( + departed, + f"'{ros_name}' still listed {DEPARTURE_TIMEOUT_SEC}s after the last cycle exited", + ) + + def test_02_a_node_with_no_parameter_services_is_still_an_app(self): + """The endpoint rule removes names that are gone, not quiet nodes.""" + publishers = self._poll_probe_publishers(SILENT_NODE_NAME) + self.assertIn( + SILENT_NODE_ENDPOINT, publishers, + f"the graph attributes no '{SILENT_NODE_ENDPOINT}' publisher to " + f"'{SILENT_NODE_NAME}' ({publishers}), so this case would pass on a " + 'gateway that lists nothing at all', + ) + + listed = self.poll_endpoint_until( + '/apps', + lambda data: data if SILENT_NODE_NAME in [ + app['id'] for app in data.get('items', []) + ] else None, + timeout=SILENT_NODE_TIMEOUT_SEC, + ) + self.assertIsNotNone( + listed, + 'a node with parameter services and the parameter-event publisher ' + f'off must still be an App; /apps held {self._app_ids()}', + ) + + +@launch_testing.post_shutdown_test() +class TestShutdown(unittest.TestCase): + """The gateway exits cleanly after a run full of departing nodes.""" + + def test_exit_codes(self, proc_info, gateway_node): + exit_code = proc_info[gateway_node].returncode + self.assertIn( + exit_code, + ALLOWED_EXIT_CODES, + f'Process gateway_node exited with {exit_code}', + ) diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/CMakeLists.txt b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/CMakeLists.txt index adc0f6463..e63632ab0 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/CMakeLists.txt +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/CMakeLists.txt @@ -49,6 +49,9 @@ set(GATEWAY_SRC_DIR "${CMAKE_CURRENT_SOURCE_DIR}/../../ros2_medkit_gateway" set(GATEWAY_SRC_INCLUDE_DIR "${GATEWAY_SRC_DIR}/include") set(LIFECYCLE_STATE_READER_SOURCES ${GATEWAY_SRC_DIR}/src/core/status/lifecycle_status_helpers.cpp + # The reader builds each call's callback group through ros2_common, which is + # where the gateway keeps node-entity creation. + ${GATEWAY_SRC_DIR}/src/ros2_common/callback_groups.cpp ${GATEWAY_SRC_DIR}/src/ros2/status/ros2_lifecycle_state_reader.cpp) # Fail early with a clear message if the reused gateway sources are missing # (e.g. building the package outside the workspace), @@ -645,6 +648,26 @@ if(BUILD_TESTING) LABELS "integration;e2e" ENV "GATEWAY_TEST_PORT=19300" "WATCHDOG_E2E_SCENARIO=restart_departed" "${_WATCHDOG_E2E_ENV}") + # === lifecycle reader contention e2e (launch_testing) === + # + # 60 (arming gate) + 15 (sampling window) plus the launch's own settle time. + # 240 to also cover teardown for TWO processes (sigterm 30 + sigkill 15 each, + # not fully serial). + medkit_add_launch_test(test_e2e_test_lifecycle_reader_contention_e2e.test.py + test/e2e/test_lifecycle_reader_contention_e2e.test.py TIMEOUT 240 + LABELS "integration;e2e" + ENV "GATEWAY_TEST_PORT=19630" "${_WATCHDOG_E2E_ENV}") + + # === lifecycle reader identity e2e (launch_testing) === + # + # 60 (arming gate) + 60 (reader visible) + 10 (settle window while both creators have + # had their chance) = 130 internal. Rounded to 240 to also cover the launch's own + # teardown for TWO processes (sigterm 30 + sigkill 15 each, not fully serial). + medkit_add_launch_test(test_e2e_test_lifecycle_reader_identity_e2e.test.py + test/e2e/test_lifecycle_reader_identity_e2e.test.py TIMEOUT 240 + LABELS "integration;e2e" + ENV "GATEWAY_TEST_PORT=19620" "${_WATCHDOG_E2E_ENV}") + # === node_death e2e (launch_testing) === # # Exercises the node_death detector end to end against a real gateway + fault_manager + diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/design/graph_watchdog.rst b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/design/graph_watchdog.rst index bba329b45..1025fc1d0 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/design/graph_watchdog.rst +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/design/graph_watchdog.rst @@ -146,6 +146,13 @@ Reliability core ``ros2_lifecycle_state_reader.cpp``) compiled in via ``GATEWAY_SRC_DIR`` - the same non-header-only reuse pattern other gateway plugins use - rather than reimplementing lifecycle-state parsing. +- **One reader per gateway.** The reader instance itself comes from the plugin + context (``RosPluginContext::lifecycle_state_reader()``), not from a + constructor call here. It owns a private ROS node named after the gateway, so + a second instance would claim a fully qualified name that is already taken - + DDS warns about it and every graph query that lists nodes returns the name + twice. ``LifecycleWatcher`` still builds its own when the context hands it + nullptr, which is what a fixture running without a gateway gets. Detectors --------- diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/lifecycle_watcher.hpp b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/lifecycle_watcher.hpp index b6eeec6bb..75fa0d087 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/lifecycle_watcher.hpp +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/lifecycle_watcher.hpp @@ -106,8 +106,14 @@ class LifecycleWatcher { /// number of attempts covers five times the wall clock at a fifth of the per-tick cost. static constexpr std::uint64_t kUnmeasuredSeedInterval = 5; + /// `reader` is the gateway's own lifecycle-state reader. It is shared rather + /// than built here because the reader owns a private ROS node named after the + /// gateway; a second one would put a second node of that name on the graph. + /// A null `reader` makes this watcher build its own, which is what a fixture + /// running without a gateway needs. LifecycleWatcher(rclcpp::Node * gateway_node, std::mutex * node_mutex, - int departed_retention_ticks = kDefaultDepartedRetentionTicks); + int departed_retention_ticks = kDefaultDepartedRetentionTicks, + std::shared_ptr reader = nullptr); ~LifecycleWatcher(); LifecycleWatcher(const LifecycleWatcher &) = delete; LifecycleWatcher & operator=(const LifecycleWatcher &) = delete; @@ -226,7 +232,7 @@ class LifecycleWatcher { }; rclcpp::Node * node_; std::mutex * node_mutex_; - std::shared_ptr reader_; + std::shared_ptr reader_; std::shared_ptr state_; int retention_ticks_; /// The lifecycle subscriptions' own callback group, created with automatic executor diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/reliability_gate.hpp b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/reliability_gate.hpp index bc2dfaaa1..6a6d7d4e3 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/reliability_gate.hpp +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/include/ros2_medkit_graph_watchdog/reliability_gate.hpp @@ -40,7 +40,8 @@ namespace ros2_medkit_graph_watchdog { class ReliabilityGate { public: ReliabilityGate(int warmup_cycles, rclcpp::Node * gateway_node, std::mutex * node_mutex, - int departed_retention_ticks = kDefaultDepartedRetentionTicks); + int departed_retention_ticks = kDefaultDepartedRetentionTicks, + std::shared_ptr reader = nullptr); /// Feed a fresh introspection snapshot at `tick`: updates per-entity warmup, /// lifecycle tracking, and the global bringup marker. diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/graph_watchdog_plugin.cpp b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/graph_watchdog_plugin.cpp index e8690bde3..b80155715 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/graph_watchdog_plugin.cpp +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/graph_watchdog_plugin.cpp @@ -266,7 +266,8 @@ void GraphWatchdogPlugin::set_context(ros2_medkit_gateway::PluginContext & conte // Computed BEFORE gate_ is constructed: node_death's own configure() (which normally // owns its miss_grace default) has not run yet at this point. const int departed_retention_ticks = compute_departed_retention_ticks(config_snapshot); - gate_ = std::make_unique(warmup_cycles_, node, &node_mutex_, departed_retention_ticks); + gate_ = std::make_unique(warmup_cycles_, node, &node_mutex_, departed_retention_ticks, + ctx_->lifecycle_state_reader()); // The fault client gets the same treatment as the lifecycle subscriptions: its own // callback group, automatic executor registration disabled, added only to a private diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/lifecycle_watcher.cpp b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/lifecycle_watcher.cpp index 9e405a1d8..c989a63ed 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/lifecycle_watcher.cpp +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/lifecycle_watcher.cpp @@ -66,10 +66,11 @@ std::string derive_transition_event_topic(const std::string & get_state_path) { } // namespace -LifecycleWatcher::LifecycleWatcher(rclcpp::Node * gateway_node, std::mutex * node_mutex, int departed_retention_ticks) +LifecycleWatcher::LifecycleWatcher(rclcpp::Node * gateway_node, std::mutex * node_mutex, int departed_retention_ticks, + std::shared_ptr reader) : node_(gateway_node) , node_mutex_(node_mutex) - , reader_(std::make_shared(gateway_node)) + , reader_(reader ? std::move(reader) : std::make_shared(gateway_node)) , state_(std::make_shared()) , retention_ticks_(departed_retention_ticks) { // One group for every lifecycle subscription, created once here rather than per node: diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/reliability_gate.cpp b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/reliability_gate.cpp index 89e8f38e0..be441a2a8 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/reliability_gate.cpp +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/src/reliability_gate.cpp @@ -16,9 +16,10 @@ namespace ros2_medkit_graph_watchdog { ReliabilityGate::ReliabilityGate(int warmup_cycles, rclcpp::Node * gateway_node, std::mutex * node_mutex, - int departed_retention_ticks) + int departed_retention_ticks, + std::shared_ptr reader) : warmup_(warmup_cycles) - , lifecycle_(gateway_node, node_mutex, departed_retention_ticks) + , lifecycle_(gateway_node, node_mutex, departed_retention_ticks, std::move(reader)) , warmup_cycles_(warmup_cycles < 0 ? 0 : warmup_cycles) { } diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/harness.py b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/harness.py index ca47df5aa..edcf95cdd 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/harness.py +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/harness.py @@ -173,6 +173,7 @@ def create_watchdog_test_launch( healing_enabled=True, healing_threshold=3, gateway_respawn=False, + gateway_name='ros2_medkit_gateway', ): """Build a ``LaunchDescription`` that loads graph_watchdog into a real gateway. @@ -209,6 +210,13 @@ def create_watchdog_test_launch( whose subject IS the debounce counter (a low threshold makes it sensitive to even a single spurious PASSED) overrides this directly rather than adding a second mechanism. + gateway_name : str or None + Name to launch the gateway under. ``None`` omits the ``__node:=`` + remap, which is the only way to see the gateway's helper nodes under + their own names: the remap is a global argument and rclcpp applies it + to every node the process creates, so under a name the gateway's + private client nodes all answer to that one name instead of their + suffixed ones. gateway_respawn : bool If True, ``launch`` restarts the gateway when it exits. For the one scenario whose subject is a gateway restart: the fault_manager and the @@ -237,7 +245,8 @@ def create_watchdog_test_launch( if extra_gateway_params: params.update(extra_gateway_params) - gateway_node = create_gateway_node(port=port, extra_params=params, respawn=gateway_respawn) + gateway_node = create_gateway_node(port=port, extra_params=params, respawn=gateway_respawn, + name=gateway_name) delayed_actions = create_demo_nodes(demo_nodes if demo_nodes is not None else []) delayed_actions.append(create_fault_manager_node( diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/test_lifecycle_reader_contention_e2e.test.py b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/test_lifecycle_reader_contention_e2e.test.py new file mode 100644 index 000000000..21c3bba99 --- /dev/null +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/test_lifecycle_reader_contention_e2e.test.py @@ -0,0 +1,195 @@ +# Copyright 2026 bburda +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +One lifecycle-state reader serves everybody, and nobody waits behind anybody. + +The gateway's ``/status`` handler and this plugin's lifecycle watcher read +managed nodes through the same reader object. That is deliberate - the reader +owns a private ROS node named after the gateway, so a second one would collide +on the graph - but it puts an HTTP request and a plugin tick on the same object, +and the plugin's seed loop talks to nodes that may never answer. + +The scenario builds exactly that: ``unreadable_lifecycle`` advertises +``get_state`` and never replies, so every seed against it spends the reader's +full timeout, and the watcher re-seeds it forever because its label never +becomes known. Meanwhile ``managed_lifecycle_active`` answers at once. A +``/status`` read of the answering node must not pay for the unanswering one. + +Numbers, all measured on this branch on a developer box, one gateway, the +launch below: + +- watcher holding its own reader, i.e. no sharing at all: max ``/status`` + latency 8 ms over 277 samples, median 4 ms +- one shared reader whose mutex spans the request and the inline spin: max + 489 ms over 190 samples, median 4 ms. That maximum is the reader's own 500 ms + timeout, spent on a node the caller never asked about +- one shared reader whose mutex covers only client creation and destruction, + five runs: maxima 6, 7, 7, 88 and 97 ms, median 4 ms every time + +The maxima in that last row are the floor this bound has to clear, and the +occasional ~90 ms one is not the reader: a read that queues behind a seed pays +the reader's whole remaining timeout, which is why the regression row is a clean +489 ms rather than a spread. 250 ms is between the two, about 2.5x the worst +measured noise and half the regression, and `get_time_scale` stretches it where +the noise is worst. + +The statistic is the maximum on purpose. The regression makes only a handful of +the samples slow - the median stays at 4 ms in the contended row above - so a +percentile rule would let it through. +""" + +import os +import statistics +import sys +import time +import unittest + +import launch_testing +import requests + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +# I100 as well as E402: `harness` is only importable because of the sys.path line above, so this +# import cannot be moved up to where the alphabetical order would put it. +from harness import ( # noqa: E402, I100 + API_BASE_PATH, + create_watchdog_test_launch, + wait_until_watchdog_armed, +) + +from ros2_medkit_test_utils.constants import ( # noqa: E402 + ALLOWED_EXIT_CODES, + get_test_port, + get_time_scale, +) + +PORT = get_test_port() + +# 200 ms ticks put an unmeasured node's re-seed every +# LifecycleWatcher::kUnmeasuredSeedInterval (5) ticks, i.e. once a second. Each +# of those seeds spends the reader's 500 ms timeout against a node that never +# replies, so roughly half of the sampling window below overlaps a seed. +TICK_INTERVAL_MS = 200 +WARMUP_CYCLES = 3 + +# Answers GetState immediately. +RESPONSIVE_APP = 'managed_lifecycle_active' +# Advertises get_state and never answers it. +UNREADABLE_APP = 'unreadable_lifecycle' + +SAMPLE_WINDOW_SEC = 15.0 * get_time_scale() +SAMPLE_INTERVAL_SEC = 0.05 +REQUEST_TIMEOUT_SEC = 10.0 * get_time_scale() + +# The reader's own default GetState timeout, which is what a caller queued +# behind one unanswering seed pays on top of its own read. +READER_TIMEOUT_SEC = 0.5 + +# Half the reader timeout: comfortably above a read that only pays for itself, +# and unreachable by one that also pays for a seed against a node that never +# answers. Scales with the sanitizer factor because every term in a served +# request does. +MAX_STATUS_LATENCY_SEC = (READER_TIMEOUT_SEC / 2) * get_time_scale() + + +def generate_test_description(): + return create_watchdog_test_launch( + detector_params={ + 'plugins.graph_watchdog.tick_interval_ms': TICK_INTERVAL_MS, + 'plugins.graph_watchdog.warmup_cycles': WARMUP_CYCLES, + }, + demo_nodes=[RESPONSIVE_APP, UNREADABLE_APP], + port=PORT, + ) + + +class TestLifecycleReaderContentionE2e(unittest.TestCase): + """A shared reader must not serialise one caller behind another's timeout.""" + + def _status_latencies(self, app_id): + url = f'http://127.0.0.1:{PORT}{API_BASE_PATH}/apps/{app_id}/status' + latencies = [] + deadline = time.monotonic() + SAMPLE_WINDOW_SEC + while time.monotonic() < deadline: + started = time.monotonic() + response = requests.get(url, timeout=REQUEST_TIMEOUT_SEC) + elapsed = time.monotonic() - started + self.assertEqual( + response.status_code, 200, + f'GET {url} returned {response.status_code}: {response.text[:200]}', + ) + latencies.append(elapsed) + time.sleep(SAMPLE_INTERVAL_SEC) + return latencies + + def test_status_does_not_queue_behind_a_watchdog_seed(self): + # Gating on the responsive app by name, not just on "some app armed": + # it is what proves the gateway has discovered it before the first + # /status read, and that the watcher is tracking lifecycle nodes. + self.assertTrue( + wait_until_watchdog_armed(PORT, app_id=RESPONSIVE_APP), + f"the watchdog never armed on '{RESPONSIVE_APP}', so its lifecycle " + 'watcher never seeded anything', + ) + self.assertTrue( + wait_until_watchdog_armed(PORT, app_id=UNREADABLE_APP), + f"the watchdog never armed on '{UNREADABLE_APP}', so nothing was " + 'contending for the reader', + ) + + # Anti-vacuity: without a responsive managed node this case would time a + # branch that never reads lifecycle state at all. + ready = requests.get( + f'http://127.0.0.1:{PORT}{API_BASE_PATH}/apps/{RESPONSIVE_APP}/status', + timeout=REQUEST_TIMEOUT_SEC, + ) + self.assertEqual(ready.status_code, 200, ready.text[:200]) + self.assertEqual( + ready.json().get('status'), 'ready', + f'{RESPONSIVE_APP} must be an active managed node for this case to ' + f'exercise the reader at all; got {ready.text[:200]}', + ) + + latencies = self._status_latencies(RESPONSIVE_APP) + self.assertGreater(len(latencies), 50, 'too few samples to say anything') + + worst = max(latencies) + # Printed on every run, not only on failure: the bound is a measurement, + # and a run that passes with no margin is worth seeing before it fails. + print( + f'/status latency over {len(latencies)} samples: ' + f'max {worst * 1000:.0f}ms, median {statistics.median(latencies) * 1000:.0f}ms, ' + f'bound {MAX_STATUS_LATENCY_SEC * 1000:.0f}ms', + flush=True, + ) + self.assertLess( + worst, MAX_STATUS_LATENCY_SEC, + f'slowest of {len(latencies)} /status reads took {worst:.3f}s ' + f'(median {statistics.median(latencies):.3f}s), over the ' + f'{MAX_STATUS_LATENCY_SEC:.3f}s bound: a read paid for the watchdog ' + f"seed against '{UNREADABLE_APP}', which never answers", + ) + + +@launch_testing.post_shutdown_test() +class TestShutdown(unittest.TestCase): + """Verify the gateway/fault_manager stack exits cleanly.""" + + def test_exit_codes(self, proc_info): + for info in proc_info: + self.assertIn( + info.returncode, + ALLOWED_EXIT_CODES, + f'Process {info.process_name} exited with {info.returncode}', + ) diff --git a/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/test_lifecycle_reader_identity_e2e.test.py b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/test_lifecycle_reader_identity_e2e.test.py new file mode 100644 index 000000000..e88bba734 --- /dev/null +++ b/src/ros2_medkit_plugins/ros2_medkit_graph_watchdog/test/e2e/test_lifecycle_reader_identity_e2e.test.py @@ -0,0 +1,138 @@ +# Copyright 2026 bburda +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +The gateway puts exactly one ``_lifecycle_state_reader`` on the graph. + +Reading a managed node's state needs a ROS node of its own, and that node is +named after the gateway. Two of them - the gateway's ``/status`` handler and +this plugin's lifecycle watcher - would therefore both claim the same fully +qualified name. ROS allows that and warns about it, and every graph query that +lists nodes then returns the name twice, which is a duplicate every reader of +the graph has to swallow. The reader is shared instead. + +Counted with rclpy rather than through the gateway: ``GET /apps`` de-duplicates +by name, so it cannot see the second one. This test only asks the graph. +""" + +import os +import sys +import time +import unittest + +import launch_testing +import rclpy + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +# I100 as well as E402: `harness` is only importable because of the sys.path line above, so this +# import cannot be moved up to where the alphabetical order would put it. +from harness import ( # noqa: E402, I100 + create_watchdog_test_launch, + wait_until_watchdog_armed, +) + +from ros2_medkit_test_utils.constants import ALLOWED_EXIT_CODES, get_test_port # noqa: E402 + +PORT = get_test_port() + +TICK_INTERVAL_MS = 200 +WARMUP_CYCLES = 3 + +READER_NODE_NAME = 'ros2_medkit_gateway_lifecycle_state_reader' + +# The probe polls until the reader is on the graph at all, then keeps reading +# for a while: the second node, when there is one, is created by the plugin +# rather than by the REST server, so the two appear at different moments and a +# single read taken too early would miss the collision. +READER_VISIBLE_TIMEOUT_SEC = 60.0 +SETTLE_SEC = 10.0 +PROBE_INTERVAL_SEC = 0.5 + + +def generate_test_description(): + return create_watchdog_test_launch( + detector_params={ + 'plugins.graph_watchdog.tick_interval_ms': TICK_INTERVAL_MS, + 'plugins.graph_watchdog.warmup_cycles': WARMUP_CYCLES, + }, + # A managed node gives the plugin's lifecycle watcher something to read, + # so the reader is exercised rather than merely constructed. + demo_nodes=['managed_lifecycle'], + port=PORT, + # Without the __node:= remap. It is a global argument, so rclcpp applies + # it to every node the gateway process creates and the reader would + # answer to the gateway's own name instead of its suffixed one. + gateway_name=None, + ) + + +class TestLifecycleReaderIdentityE2e(unittest.TestCase): + """One gateway, one lifecycle-state reader node.""" + + @classmethod + def setUpClass(cls): + rclpy.init() + cls._probe = rclpy.create_node('lifecycle_reader_identity_probe') + + @classmethod + def tearDownClass(cls): + cls._probe.destroy_node() + cls._probe = None + rclpy.shutdown() + + def _node_names(self): + return [name for name, _ in type(self)._probe.get_node_names_and_namespaces()] + + def _reader_count(self): + return sum(1 for name in self._node_names() if name == READER_NODE_NAME) + + def test_reader_node_name_is_claimed_once(self): + self.assertTrue( + wait_until_watchdog_armed(PORT), + 'the watchdog plugin never armed, so its lifecycle watcher never ran', + ) + + deadline = time.monotonic() + READER_VISIBLE_TIMEOUT_SEC + while time.monotonic() < deadline and self._reader_count() == 0: + time.sleep(PROBE_INTERVAL_SEC) + self.assertGreater( + self._reader_count(), 0, + f"'{READER_NODE_NAME}' never appeared on the graph, so this run proves " + f'nothing; the graph held {sorted(self._node_names())}', + ) + + worst = 0 + settle_deadline = time.monotonic() + SETTLE_SEC + while time.monotonic() < settle_deadline: + worst = max(worst, self._reader_count()) + time.sleep(PROBE_INTERVAL_SEC) + + self.assertEqual( + worst, 1, + f"'{READER_NODE_NAME}' was on the graph {worst} times; the gateway's own " + 'reader and the plugin watcher must share one', + ) + + +@launch_testing.post_shutdown_test() +class TestShutdown(unittest.TestCase): + """Verify the gateway/fault_manager stack exits cleanly.""" + + def test_exit_codes(self, proc_info): + for info in proc_info: + self.assertIn( + info.returncode, + ALLOWED_EXIT_CODES, + f'Process {info.process_name} exited with {info.returncode}', + )