diff --git a/src/ros2_medkit_gateway/CMakeLists.txt b/src/ros2_medkit_gateway/CMakeLists.txt index 79399e221..57001c9a5 100644 --- a/src/ros2_medkit_gateway/CMakeLists.txt +++ b/src/ros2_medkit_gateway/CMakeLists.txt @@ -550,6 +550,13 @@ if(BUILD_TESTING) target_link_libraries(test_configuration_manager gateway_ros2) medkit_set_test_domain(test_configuration_manager) + # Ros2ParameterTransport round-trip timeout coverage (#531): a + # discoverable-but-unresponsive node's parameter service must not hang the + # transport forever. + ament_add_gtest(test_ros2_parameter_transport test/test_ros2_parameter_transport.cpp) + target_link_libraries(test_ros2_parameter_transport gateway_ros2) + medkit_set_test_domain(test_ros2_parameter_transport) + find_package(std_msgs REQUIRED) # Add DiscoveryManager tests @@ -1034,6 +1041,7 @@ if(BUILD_TESTING) test_generic_client_compat test_operation_manager test_configuration_manager + test_ros2_parameter_transport test_discovery_manager test_runtime_discovery test_tls_config diff --git a/src/ros2_medkit_gateway/design/index.rst b/src/ros2_medkit_gateway/design/index.rst index a78928315..639a72bd2 100644 --- a/src/ros2_medkit_gateway/design/index.rst +++ b/src/ros2_medkit_gateway/design/index.rst @@ -561,7 +561,7 @@ Each entry below is tagged with the static library it compiles into: - Runs on configurable host and port with CORS support 5. **ConfigurationManager** ``[gateway_core]`` - Routes SOVD configuration CRUD through ``ParameterTransport`` - - Manager body is middleware-neutral; ``Ros2ParameterTransport`` adapter performs the ``rclcpp::SyncParametersClient`` calls + - Manager body is middleware-neutral; ``Ros2ParameterTransport`` adapter performs the ``rclcpp::AsyncParametersClient`` calls (explicit ``spin_until_future_complete`` so a service TIMEOUT is distinguished from a successful-but-empty response) - Lists / gets / sets individual parameter values with type conversion - Provides parameter descriptors (description, constraints, read-only flag) - The transport caches parameter clients per node for efficiency diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/configuration_manager.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/configuration_manager.hpp index 6d0a9323c..27e951905 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/configuration_manager.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/configuration_manager.hpp @@ -30,7 +30,7 @@ using json = nlohmann::json; * @brief Application service for ROS 2 node parameters. * * Pure C++; ROS-side I/O is performed by the injected ParameterTransport - * adapter (typically Ros2ParameterTransport). All rclcpp::SyncParametersClient + * adapter (typically Ros2ParameterTransport). All rclcpp::AsyncParametersClient * usage, the rclcpp::Parameter defaults cache, the negative-cache for * unreachable nodes, the spin_mutex serialising parameter-client spins, and * the JSON <-> rclcpp::ParameterValue conversion helpers live in the adapter. diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/parameter_transport.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/parameter_transport.hpp index 7aed276bc..76c92a8b2 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/parameter_transport.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/parameter_transport.hpp @@ -79,10 +79,6 @@ class ParameterTransport { /// Service availability check (same semantics as today's negative cache). virtual bool is_node_available(const std::string & node_name) const = 0; - /// Drop client/cached state for a given node (called by the manager when - /// reset / refresh is requested). - virtual void invalidate(const std::string & node_name) = 0; - /// Tear down all cached clients before rclcpp::shutdown(). Idempotent. virtual void shutdown() = 0; }; diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/transports/ros2_parameter_transport.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/transports/ros2_parameter_transport.hpp index 992ac867e..c5d60ce9c 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/transports/ros2_parameter_transport.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/transports/ros2_parameter_transport.hpp @@ -34,11 +34,18 @@ namespace ros2_medkit_gateway::ros2 { /** * @brief rclcpp adapter implementing ParameterTransport. * - * Owns the rclcpp::SyncParametersClient cache, the rclcpp::Parameter defaults + * Owns the rclcpp::AsyncParametersClient cache, the rclcpp::Parameter defaults * cache, the negative-cache for unreachable nodes, the spin_mutex serialising * parameter-client spins, and the JSON <-> rclcpp::ParameterValue conversion * helpers that previously lived inside ConfigurationManager. * + * Uses AsyncParametersClient + explicit spin_until_future_complete (not the + * SyncParametersClient) so a service TIMEOUT is distinguished from a + * successful-but-EMPTY response via rclcpp::FutureReturnCode. The sync client + * collapses SUCCESS-with-empty-result and TIMEOUT into the same empty vector, + * which caused a false-positive that negative-cached a HEALTHY node on a legit + * empty get (e.g. a param undeclared between list and get) - see #531. + * * The adapter exposes only the neutral ParameterTransport surface to the * manager; all rclcpp types stay confined to this translation unit. */ @@ -47,8 +54,8 @@ class Ros2ParameterTransport : public ParameterTransport { /** * @param node Non-owning ROS node used to declare parameters and to derive * the gateway's own FQN for the self-node short circuit. - * @param service_timeout_sec Timeout for SyncParametersClient::wait_for_service - * and underlying parameter-service calls. + * @param service_timeout_sec Timeout for AsyncParametersClient::wait_for_service + * and each spin_until_future_complete round trip. * @param negative_cache_ttl_sec How long an unavailable node remains in the * negative cache before another IPC attempt. * Set to 0 to disable. @@ -78,16 +85,30 @@ class Ros2ParameterTransport : public ParameterTransport { ParameterResult list_defaults(const std::string & node_name) override; bool is_node_available(const std::string & node_name) const override; - void invalidate(const std::string & node_name) override; void shutdown() override; private: - /// Get or create a cached SyncParametersClient for the given node. - std::shared_ptr get_param_client(const std::string & node_name); + /// Get or create a cached AsyncParametersClient for the given node. + std::shared_ptr get_param_client(const std::string & node_name); + + /// Spin param_node_ until `future` resolves or the service timeout elapses. + /// The ONLY place a parameter round trip blocks; makes TIMEOUT explicit. + /// @pre spin_mutex_ must be held by the caller (single spin on param_node_). + /// @return TIMEOUT - the node did not answer within service_timeout_sec; + /// SUCCESS - future.get() is valid (its result may legitimately be + /// an EMPTY vector: not-found / unset / raced); + /// INTERRUPTED - transport shutting down (or param_node_ gone). + template + rclcpp::FutureReturnCode spin_for(const FutureT & future); /// Cache default values for a node (called on first access). /// @pre spin_mutex_ must be held by the caller. - void cache_default_values(const std::string & node_name); + /// @return true if the node's own round-trip failed/timed out (service not + /// discoverable, or list/get IPC threw) so the caller can short-circuit + /// WITHOUT depending on the negative-cache TTL (which is a no-op when + /// negative_cache_ttl_sec_ == 0). false if defaults were cached, were + /// already cached, or the transport is shutting down. + bool cache_default_values(const std::string & node_name); /// Check if a node is in the negative cache (recently unavailable). bool is_node_unavailable(const std::string & node_name) const; @@ -136,8 +157,8 @@ class Ros2ParameterTransport : public ParameterTransport { std::map> default_values_; /// Mutex to serialize spin operations on param_node_. - /// SyncParametersClient::wait_for_service/get_parameters/etc spin param_node_ - /// internally via spin_node_until_future_complete which is NOT thread-safe. + /// spin_for() drives spin_until_future_complete on param_node_, which is NOT + /// thread-safe against a concurrent spin on the same node. /// Declared BEFORE param_node_ so it outlives the node during destruction. mutable std::timed_mutex spin_mutex_; @@ -145,9 +166,9 @@ class Ros2ParameterTransport : public ParameterTransport { /// Created once in constructor - must be in DDS graph early for fast service discovery. std::shared_ptr param_node_; - /// Cached SyncParametersClient per target node. + /// Cached AsyncParametersClient per target node. mutable std::mutex clients_mutex_; - std::map> param_clients_; + std::map> param_clients_; /// Maximum entries in negative cache before hard eviction. static constexpr size_t kMaxNegativeCacheSize = 500; diff --git a/src/ros2_medkit_gateway/src/ros2/transports/ros2_parameter_transport.cpp b/src/ros2_medkit_gateway/src/ros2/transports/ros2_parameter_transport.cpp index d44c3d6dd..8e8c820d3 100644 --- a/src/ros2_medkit_gateway/src/ros2/transports/ros2_parameter_transport.cpp +++ b/src/ros2_medkit_gateway/src/ros2/transports/ros2_parameter_transport.cpp @@ -71,21 +71,6 @@ bool Ros2ParameterTransport::is_node_available(const std::string & node_name) co return !is_node_unavailable(node_name); } -void Ros2ParameterTransport::invalidate(const std::string & node_name) { - { - std::unique_lock lock(negative_cache_mutex_); - unavailable_nodes_.erase(node_name); - } - { - std::lock_guard lock(defaults_mutex_); - default_values_.erase(node_name); - } - { - std::lock_guard lock(clients_mutex_); - param_clients_.erase(node_name); - } -} - std::optional> Ros2ParameterTransport::try_acquire_spin_lock(ParameterResult & result) { auto timeout = std::chrono::duration_cast( @@ -147,7 +132,7 @@ void Ros2ParameterTransport::mark_node_unavailable(const std::string & node_name } } -std::shared_ptr Ros2ParameterTransport::get_param_client(const std::string & node_name) { +std::shared_ptr Ros2ParameterTransport::get_param_client(const std::string & node_name) { std::lock_guard lock(clients_mutex_); auto it = param_clients_.find(node_name); if (it != param_clients_.end()) { @@ -156,11 +141,38 @@ std::shared_ptr Ros2ParameterTransport::get_param_ if (!param_node_) { return nullptr; } - auto client = std::make_shared(param_node_, node_name); + auto client = std::make_shared(param_node_, node_name); param_clients_[node_name] = client; return client; } +template +rclcpp::FutureReturnCode Ros2ParameterTransport::spin_for(const FutureT & future) { + // Copy the node shared_ptr ONCE under clients_mutex_ and spin the local copy. + // spin_mutex_ (held by the caller) serialises spins in the normal case, but + // shutdown() has a degraded path: if it cannot acquire spin_mutex_ within + // kShutdownTimeout it resets param_node_ anyway. Reading the member directly here + // would then race that reset (concurrent access to the same shared_ptr instance = + // UB). shutdown() clears param_clients_ and resets param_node_ under clients_mutex_, + // so taking clients_mutex_ synchronises the read; the local copy also keeps the node + // alive for the whole spin even if shutdown resets the member mid-call (#531). + std::shared_ptr node; + { + std::lock_guard lock(clients_mutex_); + node = param_node_; + } + // Guard against spinning a torn-down node (shutdown races with an in-flight op). + if (shutdown_requested_.load() || !node) { + return rclcpp::FutureReturnCode::INTERRUPTED; + } + // Free-function overload spins a fresh SingleThreadedExecutor over the node + // (collecting all current entities, including a just-created client) for up to + // service_timeout_sec, then tears it down - no persistent executor to reconcile + // with param_node_'s teardown. Serialised by spin_mutex_ (held by the caller), + // so only one spin runs on the node at a time. + return rclcpp::spin_until_future_complete(node, future, get_service_timeout()); +} + ParameterResult Ros2ParameterTransport::list_own_parameters() { ParameterResult result; try { @@ -257,7 +269,22 @@ ParameterResult Ros2ParameterTransport::list_parameters(const std::string & node } // Cache default values first (gives node extra time for DDS service discovery). - cache_default_values(node_name); + // This is itself a bounded round-trip. If it just failed/timed out, short-circuit + // rather than spending a SECOND up-to-service_timeout_sec_ round-trip below: + // holding spin_mutex_ across both would exceed the acquisition margin granted by + // try_acquire_spin_lock() and spuriously TIMEOUT a concurrent request to a + // different, healthy node (#531). + // + // Gate on cache_default_values' own return value, NOT is_node_unavailable(): + // the negative cache is a no-op when negative_cache_ttl_sec_ == 0 (a documented, + // in-range "disabled" value), so is_node_unavailable() would always be false there + // and the cap would silently break, reintroducing the ~2x-timeout hold. + if (cache_default_values(node_name)) { + result.success = false; + result.error_message = "Parameter service not available for node: " + node_name; + result.error_code = ParameterErrorCode::SERVICE_UNAVAILABLE; + return result; + } if (!client->wait_for_service(get_service_timeout())) { result.success = false; @@ -271,22 +298,53 @@ ParameterResult Ros2ParameterTransport::list_parameters(const std::string & node return result; } - auto param_names = client->list_parameters({}, 0); - parameters = client->get_parameters(param_names.names); - - if (parameters.empty() && !param_names.names.empty()) { - for (const auto & name : param_names.names) { - try { - auto single_params = client->get_parameters({name}); - if (!single_params.empty()) { - parameters.push_back(single_params[0]); - } - } catch (const std::exception & e) { - if (node_) { - RCLCPP_DEBUG(node_->get_logger(), "Failed to get param '%s': %s", name.c_str(), e.what()); - } - } + std::vector param_names; + { + auto list_future = client->list_parameters({}, 0); + auto rc = spin_for(list_future); + if (rc == rclcpp::FutureReturnCode::TIMEOUT) { + // The node did not answer the list request in time (explicit TIMEOUT, not + // an ambiguous empty vector). Negative-cache it and report 503 (#531). + mark_node_unavailable(node_name); + result.success = false; + result.error_message = "Parameter service did not respond for node: " + node_name; + result.error_code = ParameterErrorCode::TIMEOUT; + return result; + } + if (rc != rclcpp::FutureReturnCode::SUCCESS) { + // INTERRUPTED: transport is shutting down. Do NOT mark (node is fine). + result.success = false; + result.error_message = "Parameter query interrupted for node: " + node_name; + result.error_code = ParameterErrorCode::SHUT_DOWN; + return result; + } + param_names = list_future.get().names; + } + + // Empty names on SUCCESS = a genuine zero-parameter node -> success=[], no mark. + if (!param_names.empty()) { + auto get_future = client->get_parameters(param_names); + auto rc = spin_for(get_future); + if (rc == rclcpp::FutureReturnCode::TIMEOUT) { + mark_node_unavailable(node_name); + result.success = false; + result.error_message = "Parameter service did not respond for node: " + node_name; + result.error_code = ParameterErrorCode::TIMEOUT; + return result; + } + if (rc != rclcpp::FutureReturnCode::SUCCESS) { + result.success = false; + result.error_message = "Parameter query interrupted for node: " + node_name; + result.error_code = ParameterErrorCode::SHUT_DOWN; + return result; } + // SUCCESS: the batch result is authoritative. It may hold fewer entries than + // param_names (params raced/undeclared between list and get) or even be empty + // ("listed but not currently gettable") - that is a LEGIT response from a + // responsive node, returned as-is, NOT a timeout. No mark. The Sync-era + // per-name fallback (which existed only to disambiguate a timeout-induced + // empty batch) is gone: TIMEOUT is now explicit above (#531). + parameters = get_future.get(); } } // spin_mutex_ released - JSON building is lock-free. @@ -353,24 +411,86 @@ ParameterResult Ros2ParameterTransport::get_parameter(const std::string & node_n return result; } - auto param_names = client->list_parameters({param_name}, 1); - if (param_names.names.empty()) { + // Confirm the parameter exists via list. TIMEOUT -> node unreachable (503); + // SUCCESS with empty names -> the parameter genuinely does not exist (NOT_FOUND, + // node is responsive so NO mark). + std::vector found_names; + { + auto list_future = client->list_parameters({param_name}, 1); + auto rc = spin_for(list_future); + if (rc == rclcpp::FutureReturnCode::TIMEOUT) { + mark_node_unavailable(node_name); + result.success = false; + result.error_message = "Parameter service did not respond for node: " + node_name; + result.error_code = ParameterErrorCode::TIMEOUT; + return result; + } + if (rc != rclcpp::FutureReturnCode::SUCCESS) { + result.success = false; + result.error_message = "Parameter query interrupted for node: " + node_name; + result.error_code = ParameterErrorCode::SHUT_DOWN; + return result; + } + found_names = list_future.get().names; + } + + if (found_names.empty()) { result.success = false; result.error_message = "Parameter not found: " + param_name; result.error_code = ParameterErrorCode::NOT_FOUND; return result; } - auto parameters = client->get_parameters({param_name}); + // Read the value. TIMEOUT -> node unreachable (503) + mark. SUCCESS with an + // EMPTY result -> the node answered but has no value for this name (unset / + // raced / undeclared between list and get): that is NOT a timeout, so treat it + // as NOT_FOUND and DO NOT mark the (responsive) node. Distinguishing these two + // is the whole point of the async client - the sync client returned the same + // empty vector for both and wrongly 503'd the healthy node (#531). + std::vector parameters; + { + auto get_future = client->get_parameters({param_name}); + auto rc = spin_for(get_future); + if (rc == rclcpp::FutureReturnCode::TIMEOUT) { + mark_node_unavailable(node_name); + result.success = false; + result.error_message = "Parameter service did not respond for node: " + node_name; + result.error_code = ParameterErrorCode::TIMEOUT; + return result; + } + if (rc != rclcpp::FutureReturnCode::SUCCESS) { + result.success = false; + result.error_message = "Parameter query interrupted for node: " + node_name; + result.error_code = ParameterErrorCode::SHUT_DOWN; + return result; + } + parameters = get_future.get(); + } + if (parameters.empty()) { result.success = false; - result.error_message = "Failed to get parameter: " + param_name; - result.error_code = ParameterErrorCode::INTERNAL_ERROR; + result.error_message = "Parameter not currently set: " + param_name; + result.error_code = ParameterErrorCode::NOT_FOUND; return result; } param = parameters[0]; - descriptors = client->describe_parameters({param_name}); + + // describe_parameters is NON-ESSENTIAL (only feeds the optional description / + // read_only fields). Best-effort: only use it if the round trip SUCCEEDs; + // a TIMEOUT or interruption here just skips the descriptor - never mark, never + // fail, keep the value we already have (#531). + { + auto desc_future = client->describe_parameters({param_name}); + if (spin_for(desc_future) == rclcpp::FutureReturnCode::SUCCESS) { + descriptors = desc_future.get(); + } else if (node_) { + RCLCPP_DEBUG(node_->get_logger(), + "describe_parameters did not complete for '%s' on node '%s'; " + "returning value without descriptor", + param_name.c_str(), node_name.c_str()); + } + } } // spin_mutex_ released. json param_obj; @@ -465,31 +585,81 @@ ParameterResult Ros2ParameterTransport::set_parameter(const std::string & node_n return result; } - auto current_params = client->get_parameters({param_name}); + // Pre-read the current value purely to derive a type hint. Best-effort: a TIMEOUT + // or empty result here just leaves hint_type NOT_SET (json_to_parameter_value then + // infers from the JSON). Do NOT mark on the pre-read - it is not the authoritative + // write-timeout signal (the set below is) (#531). + std::vector current_params; + { + auto get_future = client->get_parameters({param_name}); + if (spin_for(get_future) == rclcpp::FutureReturnCode::SUCCESS) { + current_params = get_future.get(); + } else if (node_) { + RCLCPP_DEBUG(node_->get_logger(), "Pre-read of '%s' on node '%s' did not complete (continuing without hint)", + param_name.c_str(), node_name.c_str()); + } + } rclcpp::ParameterType hint_type = rclcpp::ParameterType::PARAMETER_NOT_SET; if (!current_params.empty()) { hint_type = current_params[0].get_type(); } + // json_to_parameter_value can throw on bad CLIENT input (e.g. malformed value for + // the parameter's type). It runs OUTSIDE any mark scope so a bad client value never + // negative-caches a healthy node; the outer catch maps a throw to INTERNAL_ERROR. rclcpp::ParameterValue param_value = json_to_parameter_value(value, hint_type); rclcpp::Parameter param(param_name, param_value); - auto results = client->set_parameters({param}); - if (results.empty() || !results[0].successful) { + std::vector results; + { + auto set_future = client->set_parameters({param}); + auto rc = spin_for(set_future); + if (rc == rclcpp::FutureReturnCode::TIMEOUT) { + // The node did not answer the write in time (explicit TIMEOUT). Negative-cache + // and report 503, not 500 (#531). + // + // At-least-once ambiguity: a TIMEOUT/503 here does NOT guarantee the set did NOT + // apply on the node - the request may have been delivered and executed with only + // the response lost. Callers must re-read the parameter to confirm the effective + // value rather than assuming the write was rejected. + mark_node_unavailable(node_name); + result.success = false; + result.error_message = "Parameter service did not respond for node: " + node_name; + result.error_code = ParameterErrorCode::TIMEOUT; + return result; + } + if (rc != rclcpp::FutureReturnCode::SUCCESS) { + result.success = false; + result.error_message = "Parameter set interrupted for node: " + node_name; + result.error_code = ParameterErrorCode::SHUT_DOWN; + return result; + } + results = set_future.get(); + } + + if (results.empty()) { + // SUCCESS but no result: a responsive node returns one SetParametersResult per + // param, so an empty results vector on SUCCESS is a node-side anomaly, NOT a + // timeout. Report INTERNAL_ERROR and do NOT mark the (responsive) node (#531). result.success = false; - result.error_message = results.empty() ? "Failed to set parameter" : results[0].reason; - if (!results.empty()) { - const auto & reason = results[0].reason; - if (reason.find("read-only") != std::string::npos || reason.find("read only") != std::string::npos || - reason.find("is read_only") != std::string::npos) { - result.error_code = ParameterErrorCode::READ_ONLY; - } else if (reason.find("type") != std::string::npos) { - result.error_code = ParameterErrorCode::TYPE_MISMATCH; - } else { - result.error_code = ParameterErrorCode::INVALID_VALUE; - } + result.error_message = "Set parameter returned no result for: " + param_name; + result.error_code = ParameterErrorCode::INTERNAL_ERROR; + return result; + } + + if (!results[0].successful) { + // Legitimate rejection by a RESPONSIVE node (read-only / type / invalid value): + // classify by reason, NOT a timeout. + result.success = false; + result.error_message = results[0].reason; + const auto & reason = results[0].reason; + if (reason.find("read-only") != std::string::npos || reason.find("read only") != std::string::npos || + reason.find("is read_only") != std::string::npos) { + result.error_code = ParameterErrorCode::READ_ONLY; + } else if (reason.find("type") != std::string::npos) { + result.error_code = ParameterErrorCode::TYPE_MISMATCH; } else { - result.error_code = ParameterErrorCode::INTERNAL_ERROR; + result.error_code = ParameterErrorCode::INVALID_VALUE; } return result; } @@ -801,22 +971,22 @@ rclcpp::ParameterValue Ros2ParameterTransport::json_to_parameter_value(const jso return rclcpp::ParameterValue(value.dump()); } -void Ros2ParameterTransport::cache_default_values(const std::string & node_name) { +bool Ros2ParameterTransport::cache_default_values(const std::string & node_name) { { std::lock_guard lock(defaults_mutex_); if (default_values_.find(node_name) != default_values_.end()) { - return; + return false; // already cached from a prior successful round-trip } } if (is_node_unavailable(node_name)) { - return; + return true; // already negative-cached: caller should short-circuit } try { auto client = get_param_client(node_name); if (!client) { - return; + return false; // transport shutting down, not a node-availability signal } if (!client->wait_for_service(get_service_timeout())) { @@ -825,34 +995,58 @@ void Ros2ParameterTransport::cache_default_values(const std::string & node_name) node_name.c_str()); } mark_node_unavailable(node_name); - return; + return true; // service not discoverable: the round-trip cannot proceed } - auto param_names = client->list_parameters({}, 0); + std::vector param_names; + { + auto list_future = client->list_parameters({}, 0); + auto rc = spin_for(list_future); + if (rc == rclcpp::FutureReturnCode::TIMEOUT) { + // Node did not answer: negative-cache and short-circuit. Crucially, do NOT + // cache anything - caching an empty map on a timeout would permanently poison + // default_values_ (never invalidated) so get_default/reset return NOT_FOUND + // forever even after the node recovers (#531). + mark_node_unavailable(node_name); + return true; + } + if (rc != rclcpp::FutureReturnCode::SUCCESS) { + return true; // INTERRUPTED (shutting down): short-circuit, no mark, no cache + } + param_names = list_future.get().names; + } + + if (param_names.empty()) { + // Genuine zero-parameter node (responsive, SUCCESS with empty names). This is a + // STABLE, authoritative fact, so memoize an EMPTY defaults map: get_default then + // returns NOT_FOUND (not NO_DEFAULTS_CACHED) and subsequent list/get/reset calls + // do not re-round-trip under spin_mutex_ forever. Only a TIMEOUT must avoid caching + // (poison-avoidance); a SUCCESS-empty is safe to cache (#531). + std::lock_guard lock(defaults_mutex_); + if (default_values_.find(node_name) == default_values_.end()) { + default_values_[node_name] = {}; + } + return false; + } std::vector parameters; - parameters = client->get_parameters(param_names.names); - - if (parameters.empty() && !param_names.names.empty()) { - for (const auto & name : param_names.names) { - try { - auto single_params = client->get_parameters({name}); - if (!single_params.empty()) { - parameters.push_back(single_params[0]); - } - } catch (const std::exception & e) { - // Log the per-parameter fetch failure rather than dropping it - // silently. A silent drop here surfaces later as - // NO_DEFAULTS_CACHED with no diagnostic for the operator. - if (node_) { - RCLCPP_WARN_THROTTLE(node_->get_logger(), *node_->get_clock(), 5000, - "Failed to fetch default value for parameter '%s' on node '%s': %s", name.c_str(), - node_name.c_str(), e.what()); - } - } + { + auto get_future = client->get_parameters(param_names); + auto rc = spin_for(get_future); + if (rc == rclcpp::FutureReturnCode::TIMEOUT) { + mark_node_unavailable(node_name); + return true; // timed out on the value read: do NOT cache (no empty-map poison) + } + if (rc != rclcpp::FutureReturnCode::SUCCESS) { + return true; } + parameters = get_future.get(); } + // SUCCESS with values (or a responsive node that returned an empty set - unset / + // undeclared / raced): both are authoritative, so cache the resulting map. An empty + // map here is a genuine SUCCESS-empty and is SAFE to memoize (unlike the TIMEOUT + // branches above, which must not cache to avoid permanent poison) (#531). std::map node_defaults; for (const auto & param : parameters) { node_defaults[param.get_name()] = param; @@ -865,10 +1059,17 @@ void Ros2ParameterTransport::cache_default_values(const std::string & node_name) } } } catch (const std::exception & e) { + // Defensive: no parameter round trip is expected to throw now (TIMEOUT/INTERRUPTED + // are explicit FutureReturnCodes and future.get() on SUCCESS does not throw). Any + // unexpected throw short-circuits without caching; do NOT mark (not proven a node + // timeout). if (node_) { - RCLCPP_ERROR(node_->get_logger(), "Failed to cache defaults for node '%s': %s", node_name.c_str(), e.what()); + RCLCPP_ERROR(node_->get_logger(), "Unexpected error caching defaults for node '%s': %s", node_name.c_str(), + e.what()); } + return true; } + return false; // defaults cached successfully } } // namespace ros2_medkit_gateway::ros2 diff --git a/src/ros2_medkit_gateway/test/test_configuration_manager.cpp b/src/ros2_medkit_gateway/test/test_configuration_manager.cpp index 80bc76860..e65f8769a 100644 --- a/src/ros2_medkit_gateway/test/test_configuration_manager.cpp +++ b/src/ros2_medkit_gateway/test/test_configuration_manager.cpp @@ -452,8 +452,8 @@ TEST_F(TestConfigurationManager, test_concurrent_parameter_access) { TEST_F(TestConfigurationManager, test_concurrent_parameter_operations_no_executor_error) { // Regression test: concurrent parameter operations must not cause // "Node has already been added to an executor" error. - // SyncParametersClient internally spins param_node_ - without proper - // serialization, concurrent calls would cause executor conflicts. + // spin_for() spins param_node_ (via spin_until_future_complete) - without + // proper serialization, concurrent calls would cause executor conflicts. node_->declare_parameter("concurrent_test_int", 0); node_->declare_parameter("concurrent_test_str", std::string("init")); diff --git a/src/ros2_medkit_gateway/test/test_configuration_manager_routing.cpp b/src/ros2_medkit_gateway/test/test_configuration_manager_routing.cpp index 43ba57aef..c1f4dd45c 100644 --- a/src/ros2_medkit_gateway/test/test_configuration_manager_routing.cpp +++ b/src/ros2_medkit_gateway/test/test_configuration_manager_routing.cpp @@ -159,10 +159,6 @@ class MockParameterTransport : public ParameterTransport { return true; } - void invalidate(const std::string & node_name) override { - invalidated_.push_back(node_name); - } - void shutdown() override { ++shutdown_calls_; } @@ -201,7 +197,6 @@ class MockParameterTransport : public ParameterTransport { std::string last_get_default_node_, last_get_default_name_; std::string last_list_defaults_node_; std::vector set_history_; - std::vector invalidated_; int list_calls_ = 0; int get_calls_ = 0; diff --git a/src/ros2_medkit_gateway/test/test_ros2_parameter_transport.cpp b/src/ros2_medkit_gateway/test/test_ros2_parameter_transport.cpp new file mode 100644 index 000000000..2a0e8c9a1 --- /dev/null +++ b/src/ros2_medkit_gateway/test/test_ros2_parameter_transport.cpp @@ -0,0 +1,592 @@ +// 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. + +// Regression coverage for #531: a discoverable-but-unresponsive node's +// parameter service must not be able to hang Ros2ParameterTransport forever. +// Before the fix, client->list_parameters(...)/get_parameters(...)/etc. were +// called with no timeout, so rclcpp defaulted to waiting forever once +// wait_for_service() had already succeeded (i.e. the service exists in the +// ROS graph but its callback never replies in time). That held spin_mutex_ +// and an HTTP worker forever, eventually exhausting the whole gateway thread +// pool. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include + +#include "ros2_medkit_gateway/ros2/transports/ros2_parameter_transport.hpp" + +using namespace ros2_medkit_gateway; + +namespace { + +constexpr const char * kUnresponsiveNodeName = "unresponsive_param_node"; + +} // namespace + +class TestRos2ParameterTransportUnresponsiveNode : public ::testing::Test { + protected: + static void SetUpTestSuite() { + rclcpp::init(0, nullptr); + } + + static void TearDownTestSuite() { + rclcpp::shutdown(); + } + + void SetUp() override { + // The node under test: parameter services are NOT auto-started by + // rclcpp::Node (start_parameter_services(false)); instead we hand-roll a + // ListParameters service whose callback blocks well past the transport's + // service timeout, simulating a node that IS discoverable (the service + // exists in the ROS graph, so wait_for_service() succeeds immediately) + // but never actually answers a request in time. + // + // The callback blocks on release_blocking_callback_ rather than a fixed + // sleep: it never completes on its own during the assertions below + // (proving the client-side call is bounded purely by + // Ros2ParameterTransport's own timeout, not by how long the server takes + // to reply), while still letting TearDown release it immediately so the + // test suite doesn't pay a long fixed delay every run. + rclcpp::NodeOptions unresponsive_options; + unresponsive_options.start_parameter_services(false); + unresponsive_node_ = std::make_shared(kUnresponsiveNodeName, unresponsive_options); + + list_parameters_service_ = unresponsive_node_->create_service( + "~/list_parameters", [this](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) { + while (!release_blocking_callback_.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + }); + + // The remaining parameter services just need to exist in the ROS graph + // so that wait_for_service() (which only probes service discovery, not + // responsiveness) succeeds regardless of which underlying client it + // checks. They are not expected to be invoked by the scenarios below + // since list_parameters is the first remote call in the round trip and + // it never returns within the bounded timeout. + get_parameters_service_ = unresponsive_node_->create_service( + "~/get_parameters", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + set_parameters_service_ = unresponsive_node_->create_service( + "~/set_parameters", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + describe_parameters_service_ = unresponsive_node_->create_service( + "~/describe_parameters", + [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + get_parameter_types_service_ = unresponsive_node_->create_service( + "~/get_parameter_types", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + set_parameters_atomically_service_ = + unresponsive_node_->create_service( + "~/set_parameters_atomically", + [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + + // Client-side node handed to the transport (used only for logging / the + // self-node FQN short-circuit). Ros2ParameterTransport creates and owns + // its own internal node for the actual AsyncParametersClient IPC, so this + // node does not need to be spun for parameter round trips to proceed. + client_node_ = std::make_shared("test_param_transport_client_node"); + + // Small service_timeout_sec so the test proves boundedness quickly. + // Generous negative_cache_ttl_sec so the negative-cache scenario below + // is guaranteed to hit the cache rather than re-attempting IPC. + transport_ = std::make_shared(client_node_.get(), 0.5, 60.0); + + // Spin the unresponsive node so its service callback can be dispatched + // when a request arrives. + executor_ = std::make_shared(); + executor_->add_node(unresponsive_node_); + spin_thread_running_ = true; + spin_thread_ = std::thread([this]() { + while (spin_thread_running_) { + executor_->spin_some(std::chrono::milliseconds(10)); + } + }); + + // Give the service time to register in the ROS graph before the test + // starts issuing requests (avoids a flaky wait_for_service failure that + // would mask the scenario under test). + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + } + + void TearDown() override { + // Release any in-flight blocking service callback first so the spin loop + // (spin_some, not a blocking spin()) can drain and the spin thread exit. + release_blocking_callback_ = true; + // Documented MultiThreadedExecutor teardown order: cancel -> join spin + // thread -> reset executor -> reset nodes. cancel() before join is safe + // here because the loop is a spin_some poll gated on spin_thread_running_, + // so it exits regardless; cancel() also unblocks any in-flight spin_some. + executor_->cancel(); + spin_thread_running_ = false; + if (spin_thread_.joinable()) { + spin_thread_.join(); + } + executor_->remove_node(unresponsive_node_); + transport_.reset(); + executor_.reset(); + list_parameters_service_.reset(); + get_parameters_service_.reset(); + set_parameters_service_.reset(); + describe_parameters_service_.reset(); + get_parameter_types_service_.reset(); + set_parameters_atomically_service_.reset(); + unresponsive_node_.reset(); + client_node_.reset(); + } + + std::shared_ptr unresponsive_node_; + rclcpp::Service::SharedPtr list_parameters_service_; + rclcpp::Service::SharedPtr get_parameters_service_; + rclcpp::Service::SharedPtr set_parameters_service_; + rclcpp::Service::SharedPtr describe_parameters_service_; + rclcpp::Service::SharedPtr get_parameter_types_service_; + rclcpp::Service::SharedPtr set_parameters_atomically_service_; + std::atomic release_blocking_callback_{false}; + + std::shared_ptr client_node_; + std::shared_ptr transport_; + + std::shared_ptr executor_; + std::thread spin_thread_; + std::atomic spin_thread_running_{false}; +}; + +// GREEN (post-fix) behavior: a request against a discoverable-but-never- +// responding node's parameter service returns a bounded-time error instead +// of hanging forever. +// +// RED (pre-fix) would have been: this call blocks until the service replies, +// which in this fixture is only released by TearDown - i.e. it would not +// return during the test body at all. That cannot be safely exercised in a +// test that must terminate on its own (the point of #531 is precisely that +// nothing bounds it), so instead this asserts the call completes well within +// a small multiple of service_timeout_sec (0.5s) - which is only possible +// because get_service_timeout() bounds each spin_until_future_complete round +// trip. +// +// list_parameters() also short-circuits after its internal +// cache_default_values() round-trip marks the node unavailable, so the whole +// call is bounded by ~1x service_timeout_sec (one round-trip), not ~2x (#531). +// The tighter 2s assertion documents that single-round-trip cap; the 5s bound +// is the documented outer safety limit. +TEST_F(TestRos2ParameterTransportUnresponsiveNode, ListParametersReturnsBoundedErrorInsteadOfHanging) { + auto start = std::chrono::steady_clock::now(); + auto result = transport_->list_parameters(kUnresponsiveNodeName); + auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_FALSE(result.success); + EXPECT_LT(elapsed, std::chrono::seconds(5)) + << "list_parameters() against an unresponsive-but-discoverable node must be bounded by " + "service_timeout_sec, not block until the server replies (or forever, pre-fix)"; + EXPECT_LT(elapsed, std::chrono::seconds(2)) + << "list_parameters() must cap the spin_mutex_ hold at ~1x service_timeout_sec (single " + "round-trip) by short-circuiting once cache_default_values marks the node unavailable"; +} + +// A second, immediate call against the same node must short-circuit via the +// negative cache (mark_node_unavailable() populated it on the first call's +// round-trip failure) rather than paying the timeout again. +TEST_F(TestRos2ParameterTransportUnresponsiveNode, SecondCallShortCircuitsViaNegativeCache) { + auto first_result = transport_->list_parameters(kUnresponsiveNodeName); + ASSERT_FALSE(first_result.success); + ASSERT_FALSE(transport_->is_node_available(kUnresponsiveNodeName)) + << "first round-trip failure must negative-cache the node (#531)"; + + auto start = std::chrono::steady_clock::now(); + auto second_result = transport_->list_parameters(kUnresponsiveNodeName); + auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_FALSE(second_result.success); + EXPECT_LT(elapsed, std::chrono::milliseconds(100)) + << "second call must hit the negative cache and fail fast, not re-attempt the 0.5s IPC timeout"; +} + +// A round-trip timeout against an unresponsive node must map to a +// node-unavailable error code (TIMEOUT / SERVICE_UNAVAILABLE -> HTTP 503), NOT +// INTERNAL_ERROR (-> HTTP 500). Before the fix the inner round-trip catch +// rethrew into the operation's outer catch, which set INTERNAL_ERROR and thus +// wrongly surfaced a 500 for what is really "the node isn't answering" (#531). +TEST_F(TestRos2ParameterTransportUnresponsiveNode, GetParameterTimeoutMapsTo503NotInternalError) { + auto start = std::chrono::steady_clock::now(); + auto result = transport_->get_parameter(kUnresponsiveNodeName, "some_param"); + auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_FALSE(result.success); + EXPECT_LT(elapsed, std::chrono::seconds(5)) << "get_parameter() must be bounded by service_timeout_sec"; + EXPECT_NE(result.error_code, ParameterErrorCode::INTERNAL_ERROR) + << "a node round-trip timeout must not be reported as INTERNAL_ERROR (would map to HTTP 500)"; + EXPECT_TRUE(result.error_code == ParameterErrorCode::TIMEOUT || + result.error_code == ParameterErrorCode::SERVICE_UNAVAILABLE) + << "an unresponsive node must map to a 503 (node-unavailable) error code, got " + << static_cast(result.error_code); +} + +// The list_parameters single-round-trip cap must hold even with the negative +// cache DISABLED (negative_cache_ttl_sec == 0, a documented in-range value). +// is_node_unavailable() is always false when the TTL is 0, so gating the cap on +// it would silently let list_parameters do a SECOND round-trip. The cap is +// instead driven by cache_default_values' own return value, which is +// TTL-independent (#531). +// +// Timing-INDEPENDENT assertion (does not flake under TSan/ASan, which triple +// ctest timeouts): the error CLASS is the proxy for which path ran. +// cache_default_values short-circuits with SERVICE_UNAVAILABLE, so seeing +// SERVICE_UNAVAILABLE proves the short-circuit fired (single round-trip). If the +// cap were defeated (second round-trip), list_parameters' own catch would set +// TIMEOUT instead. A generous outer time bound is kept only as a sanity net. +TEST_F(TestRos2ParameterTransportUnresponsiveNode, ListParametersCapHoldsWithNegativeCacheDisabled) { + constexpr double kServiceTimeoutSec = 1.0; + constexpr double kNegativeCacheDisabled = 0.0; + auto ttl0_transport = + std::make_shared(client_node_.get(), kServiceTimeoutSec, kNegativeCacheDisabled); + + auto start = std::chrono::steady_clock::now(); + auto result = ttl0_transport->list_parameters(kUnresponsiveNodeName); + auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_FALSE(result.success); + EXPECT_EQ(result.error_code, ParameterErrorCode::SERVICE_UNAVAILABLE) + << "with the negative cache disabled the cache_default_values short-circuit must still fire " + "(SERVICE_UNAVAILABLE); TIMEOUT would mean the cap was defeated and a second round-trip ran"; + EXPECT_LT(elapsed, std::chrono::seconds(5)) << "sanity: still bounded"; +} + +// --------------------------------------------------------------------------- +// list_parameters RESPONDS, get_parameters / set_parameters BLOCK (or answer +// empty). +// +// This is the coverage the fully-unresponsive fixture above CANNOT provide. +// The transport uses AsyncParametersClient + spin_until_future_complete, so a +// get/set that times out yields FutureReturnCode::TIMEOUT while a get/set that +// answers with an empty result yields SUCCESS - two distinct outcomes the +// (removed) SyncParametersClient collapsed into the same empty vector. A node +// that answers list (so the transport gets past the first call) but blocks the +// value read/write exercises the TIMEOUT path; the same node answering the get +// with an EMPTY value (get_returns_empty_) exercises the SUCCESS-empty path - +// which must NOT be mistaken for a timeout (#531). +// --------------------------------------------------------------------------- +class TestRos2ParameterTransportListRespondsGetSetStuck : public ::testing::Test { + protected: + static constexpr const char * kNodeName = "list_responds_getset_stuck_node"; + static constexpr const char * kParamName = "sample_param"; + + static void SetUpTestSuite() { + rclcpp::init(0, nullptr); + } + static void TearDownTestSuite() { + rclcpp::shutdown(); + } + + void SetUp() override { + rclcpp::NodeOptions options; + options.start_parameter_services(false); + node_ = std::make_shared(kNodeName, options); + + // list_parameters ALWAYS responds with one name, so the transport always + // gets past the first round-trip call and reaches get/set. + list_parameters_service_ = node_->create_service( + "~/list_parameters", [](const std::shared_ptr /*request*/, + std::shared_ptr response) { + response->result.names.push_back(kParamName); + }); + + // get_parameters: blocks while block_get_ is set (client times out -> empty). + // When get_returns_empty_ is set it RESPONDS immediately with an EMPTY values + // vector (responsive, but no value) - the crucial discriminator between a genuine + // "not currently gettable" and a timeout. Otherwise returns one INTEGER value per + // requested name (responsive). + get_parameters_service_ = node_->create_service( + "~/get_parameters", [this](const std::shared_ptr request, + std::shared_ptr response) { + while (block_get_.load() && !release_all_.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + if (release_all_.load() || get_returns_empty_.load()) { + return; // respond with an empty values vector (fast, responsive) + } + for (size_t i = 0; i < request->names.size(); ++i) { + rcl_interfaces::msg::ParameterValue value; + value.type = rcl_interfaces::msg::ParameterType::PARAMETER_INTEGER; + value.integer_value = 42; + response->values.push_back(value); + } + }); + + // set_parameters: blocks while block_set_ is set (client times out -> empty), + // otherwise returns one successful result per param (responsive). + set_parameters_service_ = node_->create_service( + "~/set_parameters", [this](const std::shared_ptr request, + std::shared_ptr response) { + while (block_set_.load() && !release_all_.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + if (release_all_.load()) { + return; + } + for (size_t i = 0; i < request->parameters.size(); ++i) { + rcl_interfaces::msg::SetParametersResult result; + result.successful = true; + response->results.push_back(result); + } + }); + + // describe_parameters / get_parameter_types / set_parameters_atomically only + // need to exist so wait_for_service() succeeds; they respond instantly empty. + describe_parameters_service_ = node_->create_service( + "~/describe_parameters", + [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + get_parameter_types_service_ = node_->create_service( + "~/get_parameter_types", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + set_parameters_atomically_service_ = node_->create_service( + "~/set_parameters_atomically", + [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + + client_node_ = std::make_shared("test_listresponds_client_node"); + + executor_ = std::make_shared(); + executor_->add_node(node_); + spin_thread_running_ = true; + spin_thread_ = std::thread([this]() { + while (spin_thread_running_) { + executor_->spin_some(std::chrono::milliseconds(10)); + } + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + } + + void TearDown() override { + release_all_ = true; // unstick any blocked callback first + executor_->cancel(); + spin_thread_running_ = false; + if (spin_thread_.joinable()) { + spin_thread_.join(); + } + executor_->remove_node(node_); + executor_.reset(); + list_parameters_service_.reset(); + get_parameters_service_.reset(); + set_parameters_service_.reset(); + describe_parameters_service_.reset(); + get_parameter_types_service_.reset(); + set_parameters_atomically_service_.reset(); + node_.reset(); + client_node_.reset(); + } + + std::shared_ptr node_; + rclcpp::Service::SharedPtr list_parameters_service_; + rclcpp::Service::SharedPtr get_parameters_service_; + rclcpp::Service::SharedPtr set_parameters_service_; + rclcpp::Service::SharedPtr describe_parameters_service_; + rclcpp::Service::SharedPtr get_parameter_types_service_; + rclcpp::Service::SharedPtr set_parameters_atomically_service_; + + std::atomic block_get_{false}; + std::atomic block_set_{false}; + std::atomic get_returns_empty_{false}; + std::atomic release_all_{false}; + + std::shared_ptr client_node_; + std::shared_ptr executor_; + std::thread spin_thread_; + std::atomic spin_thread_running_{false}; +}; + +// (a) get_parameter: list responds (param exists) but get_parameters never +// answers (blocked). spin_for returns FutureReturnCode::TIMEOUT, which must map +// to a 503-class code (TIMEOUT), NOT INTERNAL_ERROR (500), and negative-cache +// the node (#531). +TEST_F(TestRos2ParameterTransportListRespondsGetSetStuck, GetParameterEmptyGetAfterFoundNameIsTimeout) { + block_get_ = true; + auto transport = std::make_shared(client_node_.get(), 0.5, 60.0); + + auto result = transport->get_parameter(kNodeName, kParamName); + + EXPECT_FALSE(result.success); + EXPECT_EQ(result.error_code, ParameterErrorCode::TIMEOUT) + << "a blocked get_parameters after a found name is a value-read timeout, got " + << static_cast(result.error_code); + EXPECT_NE(result.error_code, ParameterErrorCode::INTERNAL_ERROR); + EXPECT_FALSE(transport->is_node_available(kNodeName)) << "the get timeout must negative-cache the node"; +} + +// (b) set_parameter: pre-read responds (fast type hint), but set_parameters +// never answers (blocked). spin_for TIMEOUT must map to TIMEOUT (503) + +// negative-cache, NOT INTERNAL_ERROR (500). This is the #531-fully-live-on-writes +// path: pre-fix a fully-unresponsive node's set returned 500 with no negative cache. +TEST_F(TestRos2ParameterTransportListRespondsGetSetStuck, SetParameterEmptyResultsIsTimeoutNotInternalError) { + block_get_ = false; // pre-read (type hint) responds quickly + block_set_ = true; // the actual write stalls + auto transport = std::make_shared(client_node_.get(), 0.5, 60.0); + + auto result = transport->set_parameter(kNodeName, kParamName, nlohmann::json(7)); + + EXPECT_FALSE(result.success); + EXPECT_EQ(result.error_code, ParameterErrorCode::TIMEOUT) + << "a blocked set_parameters is a write timeout, got " << static_cast(result.error_code); + EXPECT_NE(result.error_code, ParameterErrorCode::INTERNAL_ERROR); + EXPECT_FALSE(transport->is_node_available(kNodeName)) << "the set timeout must negative-cache the node"; +} + +// (c) list_parameters op: defaults already cached (so cache_default_values does +// not short-circuit), then get_parameters stalls on the MAIN round-trip +// (spin_for TIMEOUT). Must be reported as TIMEOUT (503), not a misleading +// success=[] (#531). Asserting TIMEOUT (not SERVICE_UNAVAILABLE) pins the main +// round-trip path specifically, distinct from the cache_default_values gate. +TEST_F(TestRos2ParameterTransportListRespondsGetSetStuck, ListParametersEmptyGetAfterNamesIsTimeoutNotEmptySuccess) { + block_get_ = false; // fully responsive first + auto transport = std::make_shared(client_node_.get(), 0.5, 60.0); + + auto first = transport->list_parameters(kNodeName); + ASSERT_TRUE(first.success) << "responsive node must list successfully and cache defaults first"; + + block_get_ = true; // now stall value reads + auto second = transport->list_parameters(kNodeName); + + EXPECT_FALSE(second.success); + EXPECT_EQ(second.error_code, ParameterErrorCode::TIMEOUT) + << "list returned names but the value read timed out: must be TIMEOUT (main round-trip), got " + << static_cast(second.error_code); +} + +// Recovery: a node negative-cached on a get timeout must recover once it becomes +// responsive again and the TTL expires - it must NOT be 503'd forever. Also +// pins Fix 4 (no empty-defaults poison): cache_default_values must NOT have +// committed an empty defaults map on the timeout, or get_default would return +// NOT_FOUND forever; after recovery get_default returns the real value. +TEST_F(TestRos2ParameterTransportListRespondsGetSetStuck, MarkedNodeRecoversAfterTtlExpires) { + constexpr double kServiceTimeoutSec = 0.5; + constexpr double kShortTtlSec = 0.4; + auto transport = std::make_shared(client_node_.get(), kServiceTimeoutSec, kShortTtlSec); + + block_get_ = true; + auto stalled = transport->list_parameters(kNodeName); + EXPECT_FALSE(stalled.success); + EXPECT_FALSE(transport->is_node_available(kNodeName)) << "get timeout must negative-cache the node"; + + // Node becomes responsive again; wait past the negative-cache TTL. + block_get_ = false; + std::this_thread::sleep_for(std::chrono::milliseconds(700)); + + auto recovered = transport->list_parameters(kNodeName); + EXPECT_TRUE(recovered.success) << "a marked node must recover after the TTL expires, not stay 503 forever"; + + // Fix 4: defaults were NOT poisoned with an empty map on the earlier timeout, + // so after recovery the real default is retrievable (not NOT_FOUND forever). + auto def = transport->get_default(kNodeName, kParamName); + EXPECT_TRUE(def.success) << "get_default must return the real default after recovery, not a poisoned empty cache"; +} + +// --------------------------------------------------------------------------- +// THE discriminating test (the async refactor's reason to exist). +// +// A node whose get_parameters service RESPONDS but returns an EMPTY values +// vector (responsive, NOT blocked - e.g. the param was undeclared/raced between +// list and get) must be treated as "not found", NOT as a timeout. The node is +// HEALTHY and must stay available (never negative-cached / 503'd). +// +// The pre-async (SyncParametersClient) code CANNOT tell this responsive-empty +// apart from a timeout - both collapse to the same empty vector - so it wrongly +// marks the node and returns TIMEOUT/503. This test FAILS against that code +// (base=RED) and PASSES with the AsyncParametersClient + FutureReturnCode +// refactor, where SUCCESS-with-empty is distinct from TIMEOUT (#531). +// --------------------------------------------------------------------------- +TEST_F(TestRos2ParameterTransportListRespondsGetSetStuck, GetParameterResponsiveEmptyIsNotFoundNotMarked) { + block_get_ = false; // do NOT block: the node answers promptly... + get_returns_empty_ = true; // ...but with an empty values vector (responsive) + auto transport = std::make_shared(client_node_.get(), 0.5, 60.0); + + auto result = transport->get_parameter(kNodeName, kParamName); + + EXPECT_FALSE(result.success); + EXPECT_EQ(result.error_code, ParameterErrorCode::NOT_FOUND) + << "a responsive node that answers get with an empty value is NOT_FOUND, not a timeout; got " + << static_cast(result.error_code); + EXPECT_NE(result.error_code, ParameterErrorCode::TIMEOUT); + EXPECT_TRUE(transport->is_node_available(kNodeName)) + << "a responsive-but-empty get must NOT negative-cache the healthy node (the pre-async " + "code wrongly marks it here)"; +} + +// The list op form of the same discriminator: a responsive node that answers get +// with an empty values vector must yield success=[] with the node still available, +// NOT a 503 / negative-cache. Pre-async this responsive-empty was marked + 503'd. +TEST_F(TestRos2ParameterTransportListRespondsGetSetStuck, ListParametersResponsiveEmptyGetIsEmptySuccessNotMarked) { + block_get_ = false; + get_returns_empty_ = true; + auto transport = std::make_shared(client_node_.get(), 0.5, 60.0); + + auto result = transport->list_parameters(kNodeName); + + EXPECT_TRUE(result.success) << "responsive node (empty get) must return success, not a 503; error_code=" + << static_cast(result.error_code); + EXPECT_TRUE(result.data.is_array()); + EXPECT_TRUE(result.data.empty()) << "no gettable values -> empty parameter list"; + EXPECT_TRUE(transport->is_node_available(kNodeName)) << "a responsive-but-empty get must NOT negative-cache the node"; +} + +// A genuine SUCCESS-empty (responsive node, empty defaults) is a STABLE, authoritative +// fact and must be MEMOIZED, so: +// - get_default returns NOT_FOUND (empty map cached), NOT NO_DEFAULTS_CACHED, and +// - the second call answers from the cache WITHOUT another round trip. +// Only a TIMEOUT must avoid caching (poison-avoidance). This is the polish that restores +// memoization the async refactor had over-broadly dropped (#531). +TEST_F(TestRos2ParameterTransportListRespondsGetSetStuck, ResponsiveEmptyMemoizesEmptyDefaults) { + get_returns_empty_ = true; // responsive node whose get returns an empty value set + auto transport = std::make_shared(client_node_.get(), 0.5, 60.0); + + auto listed = transport->list_parameters(kNodeName); + ASSERT_TRUE(listed.success); + ASSERT_TRUE(transport->is_node_available(kNodeName)); + + // The empty defaults map was cached, so get_default is NOT_FOUND (not NO_DEFAULTS_CACHED). + auto def = transport->get_default(kNodeName, kParamName); + EXPECT_FALSE(def.success); + EXPECT_EQ(def.error_code, ParameterErrorCode::NOT_FOUND) + << "a genuine SUCCESS-empty must memoize an empty defaults map (NOT_FOUND), not leave it " + "uncached (NO_DEFAULTS_CACHED); got " + << static_cast(def.error_code); + + // Block the node: a cached map means get_default must NOT re-round-trip (which would now + // time out) - it answers from the cache, so still NOT_FOUND and the node stays available. + block_get_ = true; + auto def_cached = transport->get_default(kNodeName, kParamName); + EXPECT_FALSE(def_cached.success); + EXPECT_EQ(def_cached.error_code, ParameterErrorCode::NOT_FOUND) + << "get_default must answer from the cached empty map, not re-round-trip; got " + << static_cast(def_cached.error_code); + EXPECT_TRUE(transport->is_node_available(kNodeName)) << "no re-round-trip means the node is never marked"; +} diff --git a/src/ros2_medkit_integration_tests/CMakeLists.txt b/src/ros2_medkit_integration_tests/CMakeLists.txt index 7e07df706..c5836de82 100644 --- a/src/ros2_medkit_integration_tests/CMakeLists.txt +++ b/src/ros2_medkit_integration_tests/CMakeLists.txt @@ -76,6 +76,10 @@ medkit_target_dependencies(demo_param_beacon_node rclcpp) add_executable(managed_lifecycle demo_nodes/managed_lifecycle_node.cpp) medkit_target_dependencies(managed_lifecycle rclcpp rclcpp_lifecycle) +add_executable(demo_unresponsive_param_node demo_nodes/unresponsive_param_node.cpp) +target_include_directories(demo_unresponsive_param_node PRIVATE ${_demo_include_dir}) +medkit_target_dependencies(demo_unresponsive_param_node rclcpp rcl_interfaces) + install(TARGETS demo_engine_temp_sensor demo_rpm_sensor @@ -89,6 +93,7 @@ install(TARGETS demo_beacon_publisher demo_param_beacon_node managed_lifecycle + demo_unresponsive_param_node DESTINATION lib/${PROJECT_NAME} ) diff --git a/src/ros2_medkit_integration_tests/demo_nodes/unresponsive_param_node.cpp b/src/ros2_medkit_integration_tests/demo_nodes/unresponsive_param_node.cpp new file mode 100644 index 000000000..74e12590c --- /dev/null +++ b/src/ros2_medkit_integration_tests/demo_nodes/unresponsive_param_node.cpp @@ -0,0 +1,190 @@ +// 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. + +/** + * @file unresponsive_param_node.cpp + * @brief Demo node whose parameter service is discoverable but never replies + * + * Regression fixture for #531 (end-to-end counterpart of + * ros2_medkit_gateway/test/test_ros2_parameter_transport.cpp). Parameter + * services are NOT auto-started (`start_parameter_services(false)`); + * instead all six are created manually so that a `SyncParametersClient`'s + * `wait_for_service()` succeeds immediately (the services genuinely exist + * in the ROS graph), while `~/list_parameters` - the first round-trip call + * the gateway makes - never returns. + * + * Used by test_param_service_hang.test.py to prove the gateway's REST API + * stays responsive - bounded 503 errors instead of hanging forever - when a + * backing node's parameter service is discoverable but unresponsive. + */ + +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include + +class UnresponsiveParamNode : public rclcpp::Node { + public: + UnresponsiveParamNode() : Node("unresponsive_param_node", rclcpp::NodeOptions().start_parameter_services(false)) { + list_parameters_service_ = create_service( + "~/list_parameters", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) { + // Never actually replies. Loops on rclcpp::ok() (not a fixed + // sleep) rather than a condition variable tied to node state, so + // it needs no member access at all: main() flips rclcpp::ok() to + // false on SIGINT/SIGTERM before asking the executor to stop, so + // this callback - and therefore the process - still exits + // promptly instead of hanging the executor forever. + while (rclcpp::ok()) { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + }); + + // The remaining parameter services only need to exist in the ROS graph + // so that wait_for_service() (discovery only, not responsiveness) + // succeeds for every SyncParametersClient call the gateway might issue. + // They are not expected to be invoked in practice: list_parameters is + // the first round-trip call in the gateway's configuration round trip + // and it never returns within the gateway's bounded timeout. + get_parameters_service_ = create_service( + "~/get_parameters", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + set_parameters_service_ = create_service( + "~/set_parameters", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + describe_parameters_service_ = create_service( + "~/describe_parameters", + [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + get_parameter_types_service_ = create_service( + "~/get_parameter_types", [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + set_parameters_atomically_service_ = create_service( + "~/set_parameters_atomically", + [](const std::shared_ptr /*request*/, + std::shared_ptr /*response*/) {}); + + RCLCPP_INFO(get_logger(), + "UnresponsiveParamNode started: parameter services are registered but " + "list_parameters will never reply (regression fixture for #531)"); + } + + ~UnresponsiveParamNode() override { + list_parameters_service_.reset(); + get_parameters_service_.reset(); + set_parameters_service_.reset(); + describe_parameters_service_.reset(); + get_parameter_types_service_.reset(); + set_parameters_atomically_service_.reset(); + } + + UnresponsiveParamNode(const UnresponsiveParamNode &) = delete; + UnresponsiveParamNode & operator=(const UnresponsiveParamNode &) = delete; + UnresponsiveParamNode(UnresponsiveParamNode &&) = delete; + UnresponsiveParamNode & operator=(UnresponsiveParamNode &&) = delete; + + private: + rclcpp::Service::SharedPtr list_parameters_service_; + rclcpp::Service::SharedPtr get_parameters_service_; + rclcpp::Service::SharedPtr set_parameters_service_; + rclcpp::Service::SharedPtr describe_parameters_service_; + rclcpp::Service::SharedPtr get_parameter_types_service_; + rclcpp::Service::SharedPtr set_parameters_atomically_service_; +}; + +int main(int argc, char ** argv) { + // Deliberately NOT ros2_medkit_integration_tests::run_demo_node(): that + // helper calls rclcpp::shutdown() only after executor.spin() has already + // returned, which works for every other demo node but would deadlock this + // one - the list_parameters callback above blocks the executor's worker on + // rclcpp::ok(), so shutdown() must be called BEFORE the executor is asked + // to stop, not after, or spin() never returns. + // + // Same building block as run_demo_node() otherwise: block SIGINT/SIGTERM + // up front and consume them on a dedicated sigwait thread (a normal thread + // context) rather than rclcpp's own async-signal-context handler, which is + // what makes it safe to call rclcpp::shutdown() from that thread at all. + sigset_t mask; + sigset_t old_mask; + sigemptyset(&mask); + sigaddset(&mask, SIGINT); + sigaddset(&mask, SIGTERM); + if (pthread_sigmask(SIG_BLOCK, &mask, &old_mask) != 0) { + return 1; + } + + rclcpp::init(argc, argv, rclcpp::InitOptions(), rclcpp::SignalHandlerOptions::None); + + { + auto node = std::make_shared(); + + // MultiThreadedExecutor is not strictly required: the gateway's + // Ros2ParameterTransport serializes all parameter round trips behind a + // single process-wide spin_mutex_, so only one ~/list_parameters RPC is + // ever in flight at this node at a time and a SingleThreadedExecutor + // would behave identically. The concurrency the /health test stresses + // lives in the gateway (std::async fan-out contending on that + // spin_mutex_), not in ROS-service dispatch here. Kept multi-threaded + // only as harmless future-proofing. + rclcpp::executors::MultiThreadedExecutor executor; + executor.add_node(node); + + std::thread signal_thread([&mask, &executor] { + int signum = 0; + while (true) { + const int rc = sigwait(&mask, &signum); + if (rc == 0) { + break; + } + // EINTR can theoretically fire; retry. + } + // DELIBERATE ORDER - shutdown() BEFORE cancel(), the reverse of the + // standard demo-node teardown (cancel -> join -> reset). The + // ~/list_parameters callback above blocks its executor worker in + // `while (rclcpp::ok())`, so the usual cancel()-first pattern would + // deadlock here: cancel() does not interrupt a callback that is already + // running, ok() would still be true, the callback would never return, + // and spin()/cancel()/join would all hang (forcing a SIGKILL + // escalation). Calling rclcpp::shutdown() first flips ok() to false, + // which unblocks the callback so it returns; only then is it safe to + // cancel() the (now idle) executor. This is why the post-shutdown e2e + // check sees clean exit codes instead of a SIGKILL. + rclcpp::shutdown(); + executor.cancel(); + }); + + executor.spin(); + + if (signal_thread.joinable()) { + signal_thread.join(); + } + + executor.remove_node(node); + node.reset(); + } + + // rclcpp::shutdown() already ran on the signal thread above. + pthread_sigmask(SIG_SETMASK, &old_mask, nullptr); + return 0; +} 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 3b373a0c4..ae36a297b 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 @@ -59,6 +59,9 @@ # the "active" lifecycle state (-> status "ready"). Distinct node name so it # can run alongside the unconfigured 'managed_lifecycle' in the same test. 'managed_lifecycle_active': ('managed_lifecycle', 'managed_lifecycle_active', ''), + # Regression fixture (#531): parameter services are discoverable + # (wait_for_service succeeds) but list_parameters never replies. + 'unresponsive_param': ('demo_unresponsive_param_node', 'unresponsive_param', ''), } # Convenience groupings for callers that want subsets of demo nodes. diff --git a/src/ros2_medkit_integration_tests/test/features/test_param_service_hang.test.py b/src/ros2_medkit_integration_tests/test/features/test_param_service_hang.test.py new file mode 100644 index 000000000..a247d92b2 --- /dev/null +++ b/src/ros2_medkit_integration_tests/test/features/test_param_service_hang.test.py @@ -0,0 +1,262 @@ +#!/usr/bin/env python3 +# 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. + +"""End-to-end regression coverage for #531. + +A discoverable-but-unresponsive node's parameter service must not be able to +hang the gateway's REST API. + +Unit/component coverage already proves Ros2ParameterTransport itself bounds +every SyncParametersClient call (see +ros2_medkit_gateway/test/test_ros2_parameter_transport.cpp). What is proven +here is the actual user-facing symptom against a real gateway process talking +to a real (discoverable but never-replying) demo node over the ROS 2 graph. + +The REAL gate is ``test_01_fresh_configurations_request_is_bounded_not_hung``. +It MUST run first, on a cold Ros2ParameterTransport: the shared ``spin_mutex_`` +is free and this node is not yet negative-cached, so a single ``/configurations`` +GET grabs the free lock and, pre-fix (SyncParametersClient calls with no +timeout), blocks FOREVER inside cache_default_values waiting for a reply the +node never sends - nothing else holds the lock, so there is no acquisition- +timeout rescue and the request simply never returns. The client-side timeout +then fires and fails the test (RED). Post-fix the round trip is bounded by +``parameter_service_timeout_sec`` and the gateway returns a 503 in ~2s (GREEN). +This test was verified RED against the unfixed transport (timeouts removed) and +GREEN with the fix. + +``test_02_health_stays_responsive_under_concurrent_config_calls`` is a +SECONDARY smoke test, deliberately NOT the hang gate. Because the shared +``spin_mutex_`` already serialised parameter round trips even pre-fix, ``/health`` +stays responsive under concurrent single-node ``/configurations`` calls in BOTH +the fixed and unfixed builds (only one worker ever wedges; the rest time out on +lock acquisition and free up), so this check cannot distinguish fixed from +broken - test_01 does that. By the time test_02 runs, test_01 has already +negative-cached the node, so its concurrent calls return fast via that cache; +what it asserts is only the weaker property that the REST API keeps serving +unrelated endpoints while several parameter calls are in flight. +""" + +import threading +import time +import unittest + +import launch_testing +import launch_testing.actions +import requests + +from ros2_medkit_test_utils.constants import ALLOWED_EXIT_CODES +from ros2_medkit_test_utils.gateway_test_case import GatewayTestCase +from ros2_medkit_test_utils.launch_helpers import create_test_launch + +APP_ID = 'unresponsive_param' + +# More concurrent config calls than server.http_thread_pool_size (6, see +# gateway_params.yaml). Post test_01 these are fast negative-cache hits, so this +# no longer pins every worker - it just exercises concurrent request handling. +CONCURRENT_CONFIG_REQUESTS = 8 + + +def generate_test_description(): + return create_test_launch( + demo_nodes=[APP_ID], + fault_manager=False, + ) + + +class TestParamServiceHang(GatewayTestCase): + """Proves an unresponsive parameter service cannot hang the REST API.""" + + MIN_EXPECTED_APPS = 1 + REQUIRED_APPS = {APP_ID} + + def test_01_fresh_configurations_request_is_bounded_not_hung(self): + """A single FRESH /configurations GET must return a bounded 503, not hang. + + THIS is the #531 gate (see module docstring). It runs first (the + ``test_01_`` prefix orders it ahead of the smoke test), so the + gateway's Ros2ParameterTransport is in its initial state for this + node: the shared spin_mutex_ is free and the node is not yet + negative-cached. + + Pre-fix, the underlying SyncParametersClient calls carried NO timeout, + so this one fresh request grabs the free lock, enters + cache_default_values -> list_parameters, and blocks FOREVER waiting + for a reply the unresponsive node never sends. Nothing else holds the + lock, so there is no acquisition-timeout rescue - the request never + returns, the client-side timeout below fires, and the test fails (RED). + + Post-fix, every SyncParametersClient call is bounded by + parameter_service_timeout_sec (default 2.0s): cache_default_values' + round trip times out, the node is negative-cached, and list_parameters + short-circuits, so the gateway returns a 503 SOVD error in ~2s (GREEN). + + Running first also means THIS request is what negative-caches the node + for the following smoke test. + """ + try: + start = time.monotonic() + response = requests.get( + f'{self.BASE_URL}/apps/{APP_ID}/configurations', + timeout=15, + ) + elapsed = time.monotonic() - start + except requests.exceptions.RequestException as e: + self.fail( + 'GET /configurations against a fresh unresponsive node did ' + f'not return within the 15s client timeout ({e}) - the gateway ' + 'is not bounding the parameter round trip: the request hung ' + 'instead of returning a 503 (#531 regression).' + ) + + print( + f'GET /apps/{APP_ID}/configurations returned ' + f'{response.status_code} in {elapsed:.2f}s (fresh, uncontended)' + ) + self.assertEqual( + response.status_code, 503, + f'expected 503 (node unavailable), got {response.status_code}' + ) + # Bounded well under the 15s client timeout: a fresh round trip is + # ~parameter_service_timeout_sec (2s); 10s leaves generous slack for + # sanitizer CI without letting a near-hang masquerade as a pass. + self.assertLess( + elapsed, 10.0, + 'fresh /configurations against an unresponsive node took ' + f'{elapsed:.1f}s - expected a bounded failure within a small ' + 'multiple of parameter_service_timeout_sec, not a near-timeout ' + 'rescue by the test client itself' + ) + # x-medkit-* vendor codes are remapped to the SOVD vendor-error + # envelope on the wire (see write_generic_error in + # http/detail/primitives.cpp): error_code stays a generic SOVD code, + # the precise vendor code moves to vendor_code. + data = response.json() + self.assertEqual(data.get('error_code'), 'vendor-error') + self.assertEqual(data.get('vendor_code'), 'x-medkit-ros2-node-unavailable') + self.assertIn('entity_id', data.get('parameters', {})) + self.assertEqual(data['parameters']['entity_id'], APP_ID) + + def test_02_health_stays_responsive_under_concurrent_config_calls(self): + """SMOKE: /health keeps answering with several /configurations in flight. + + Secondary coverage, NOT the #531 gate (test_01 is - see module + docstring). This does not and cannot prove "no worker ever frees": + even pre-fix the shared spin_mutex_ serialises parameter round trips, + so under N concurrent single-node /configurations only one worker + wedges while the rest time out on lock acquisition and free up, and + /health is served either way. It asserts only the weaker, still-useful + property that the REST API stays responsive while several parameter + calls are in flight. + + By the time this runs, test_01 has already negative-cached the node, + so these concurrent calls return fast via that cache rather than + blocking on fresh round trips - which is itself end-to-end proof the + negative cache works, just not a hang gate. + """ + config_threads = [] + # +1 party for this (main) thread: all hammer threads block at the + # barrier until every one has started, then release together, so the + # concurrent requests actually overlap rather than trickling out under + # scheduler-dependent thread startup. + release_barrier = threading.Barrier(CONCURRENT_CONFIG_REQUESTS + 1) + + def hammer_configurations(): + release_barrier.wait() + try: + requests.get( + f'{self.BASE_URL}/apps/{APP_ID}/configurations', + timeout=20, + ) + except requests.exceptions.RequestException: + # These threads exist only to put concurrent requests in + # flight; test_01 already asserted the response shape. + pass + + for _ in range(CONCURRENT_CONFIG_REQUESTS): + thread = threading.Thread(target=hammer_configurations, daemon=True) + config_threads.append(thread) + thread.start() + + try: + # Release all hammer threads to fire their requests together. + release_barrier.wait(timeout=10) + + # Small fixed buffer so the just-released requests actually reach + # the server and occupy workers before probing /health, rather than + # /health racing them to a free worker. + time.sleep(0.3) + + start = time.monotonic() + response = requests.get(f'{self.BASE_URL}/health', timeout=30) + elapsed = time.monotonic() - start + except threading.BrokenBarrierError: + # release_barrier.wait(timeout=10) tripped its own timeout (or the + # barrier was otherwise broken): not all hammer threads reached the + # rendezvous. Turn this into a clear failure instead of an opaque + # BrokenBarrierError traceback that would ERROR the test. + self.fail( + f'hammer threads did not reach the barrier within 10s ' + f'(expected {CONCURRENT_CONFIG_REQUESTS} + 1 parties) - could ' + 'not put the concurrent /configurations calls in flight' + ) + except requests.exceptions.RequestException as e: + self.fail( + f'/health did not respond within 30s while ' + f'{CONCURRENT_CONFIG_REQUESTS} concurrent /configurations ' + f'calls were in flight against the unresponsive node: {e}' + ) + finally: + for thread in config_threads: + thread.join(timeout=25) + + print( + f'GET /health returned {response.status_code} in {elapsed:.2f}s ' + f'while {CONCURRENT_CONFIG_REQUESTS} concurrent /configurations ' + 'calls were in flight against the unresponsive node' + ) + self.assertEqual( + response.status_code, 200, + f'/health returned {response.status_code} while ' + f'{CONCURRENT_CONFIG_REQUESTS} concurrent /configurations calls ' + 'were in flight' + ) + # Generous 30s bound: removes sanitizer-CI flake risk (TSan slows this + # thread/lock-heavy path) while still comfortably under the 120s CTest + # TIMEOUT. This is a responsiveness smoke check, not the hang gate. + self.assertLess( + elapsed, 30.0, + f'/health took {elapsed:.1f}s to respond while ' + f'{CONCURRENT_CONFIG_REQUESTS} concurrent /configurations calls ' + 'were in flight - the REST API must stay responsive' + ) + + +@launch_testing.post_shutdown_test() +class TestShutdown(unittest.TestCase): + + def test_exit_codes(self, proc_info): + """Check all processes exited cleanly (SIGTERM allowed). + + Covers the demo node's own shutdown path too: its list_parameters + callback blocks the executor on rclcpp::ok(), so this also proves + the node's custom SIGTERM handling (see unresponsive_param_node.cpp) + actually unblocks it and exits cleanly instead of requiring a + SIGKILL escalation. + """ + for info in proc_info: + self.assertIn( + info.returncode, ALLOWED_EXIT_CODES, + f'{info.process_name} exited with code {info.returncode}' + )