fix(tests,ci): stabilize shutdown-race flakes across distros

Integration-test CI has been flaky on Humble and Jazzy with a rotating
victim (demo_calibration_service, demo_brake_actuator,
demo_brake_pressure_sensor, demo_engine_temp_sensor, demo_lidar_sensor,
fault_manager_node) dying during shutdown and failing test_exit_codes.
Prior commits in this area (3adf71b, 1dc0a7e, 2f23c09 and others)
added defensive cleanup to individual demo nodes; the crashes kept
moving which was a sign the root cause was not in the nodes. This
change addresses the six distinct bugs that were actually responsible.

1. SIGINT-during-rclcpp-init race in all demo nodes
-----------------------------------------------------
If SIGINT fires between rclcpp::init() and the executor's guard
condition being allocated, rclcpp's default signal handler
invalidates the shared context mid-call; in-flight rcl_service_init /
rcl_guard_condition_init throws RCLError, the exception escapes the
bare demo main() and std::terminate aborts the process with SIGABRT.

Fix: block SIGINT/SIGTERM with pthread_sigmask, let rclcpp::init()
install its handler, construct the node, add it to an explicit
SingleThreadedExecutor (so the guard condition is allocated while
signals are still blocked), then restore the mask. A signal queued in
that window is delivered to a fully-assembled executor and shuts the
node down normally. Applied uniformly to all 10 demo nodes. The
pre-existing std::set_terminate in long_calibration_action.cpp
documents a separate rclcpp_action mid-goal-shutdown race and is
kept.

Local repro on Humble: 3 crashes / 180 iters unpatched vs 0 / 180
patched, in the 50-200 ms post-launch SIGINT window.

2. FastRTPS discovery-teardown use-after-free
-----------------------------------------------
Humble ships FastRTPS 2.6 which has a use-after-free in its
endpoint-discovery-protocol proxy tables on peer disappearance.
When the gateway shuts down cleanly, its "writer leaving" messages
are processed by peer nodes' FastRTPS listener thread which segfaults
in EDP::unpairWriterProxy / PDP::removeWriterProxyData. Backtrace
captured under stress via LD_PRELOAD=libSegFault.so confirms the
crash is in libfastrtps.so, not in our code.

Fix: set RMW_IMPLEMENTATION=rmw_cyclonedds_cpp in the CI test run
environment (both build-and-test matrix covering humble/rolling and
the jazzy-test job). CycloneDDS has a more robust discovery teardown
path.

Local repro: 1 crash / 6 iters with FastRTPS, 0 / 30 with CycloneDDS.
Also install ros-${distro}-rmw-cyclonedds-cpp in the deps step.

3. Unhandled context-invalid throws in fault_manager
------------------------------------------------------
RosbagCapture::get_topic_type, RosbagCapture::resolve_topics, and
SnapshotCapture::get_topic_type invoke node_->get_topic_names_and_types
which throws std::runtime_error when the rcl context is invalidated
mid-call. In RosbagCapture the call is reachable from a discovery
retry timer that fires every 500 ms, and the race window widens under
sanitizer load to the point where the TSan CI job hits it reliably
(SIGABRT / exit -6).

The gateway's cache refresh already catches and logs this
gracefully. Apply the same approach at the three fault_manager call
sites: wrap the rcl call in try/catch and treat invalid-context as
"no topic info available right now" - callers already handle the
empty return (skip subscribe, skip snapshot, retry next tick).

4. TSan-flagged race on RosbagCapture::post_fault_timer_
----------------------------------------------------------
TSan flags a data race on the post_fault_timer_ member
(rclcpp::TimerBase::SharedPtr):

  WRITE  on_fault_confirmed() at rosbag_capture.cpp:208
         (service callback thread)
  READ   post_fault_timer_callback() at rosbag_capture.cpp:656/658
         (main executor thread)

Both paths touch the SharedPtr without synchronisation. Even outside
TSan this is a real use-after-free waiting for the right scheduling.

Fix: introduce std::mutex post_fault_timer_mutex_ and guard every
access (on_fault_confirmed, post_fault_timer_callback, stop). Scope
kept minimal so the executor is not blocked while the service callback
creates the timer.

5. REST server start/stop race in GatewayNode
-----------------------------------------------
GatewayNode::start_rest_server() signalled "server running" on the
worker thread before calling rest_server_->start() (which is where
cpp-httplib's listen() actually binds and enters the accept loop).
For short-lived test fixtures that construct and immediately destroy
a GatewayNode (e.g. the per-test SetUp/TearDown in
test_plugin_notify_integration), the main thread can race ahead and
call stop_rest_server() before the worker has reached listen(). In
that order the stop() request is dropped, listen() subsequently
blocks forever, and the destructor's condition-variable wait hangs
until the 60s test timeout fires.

Fix: drop the server_running_/server_cv_/server_mutex_ machinery and
poll rest_server_->is_running() (which reflects cpp-httplib's actual
accept-loop state). start_rest_server now returns only once the
server is observably ready; stop_rest_server stops and joins directly.
Added RESTServer::is_running() as a passthrough to HttpServerManager.

6. test_cross_ecu_fanout poll accepts pre-discovery responses
---------------------------------------------------------------
test_data_include_peer_topics used `lambda d: d if d.get('items')`
as its readiness predicate. The gateway's /data response contains a
/rosout item as soon as LogManager subscribes, well before demo-node
discovery completes; the poll therefore returned immediately with only
{topic: '/rosout'} and the subsequent assertTrue(has_local) failed.
Tighten the predicate to require both a /powertrain/engine/ item and
a /chassis/brakes/ or /perception/lidar/ item before returning.

Verification
------------
Local stress (Humble container under 16-CPU stress-ng pressure,
LD_PRELOAD=libSegFault.so, RMW_IMPLEMENTATION=rmw_cyclonedds_cpp):
  - 200 iters of test_entity_listing: 0 fails
  -  30 iters of test_triggers_updates: 0 fails
  - fault_manager unit + integration tests: pass
  - test_plugin_notify_integration (Jazzy): 5/5 subtests pass
  - test_cross_ecu_fanout (Jazzy): pass

Author: bburda

.github/workflows/ci.yml

@@ -54,6 +54,9 @@ jobs:
         run: |
           apt-get update
           apt-get install -y ros-${{ matrix.ros_distro }}-test-msgs
+          if [ "${{ matrix.ros_distro }}" = "humble" ]; then
+            apt-get install -y ros-humble-rmw-cyclonedds-cpp
+          fi
           source /opt/ros/${{ matrix.ros_distro }}/setup.bash
           rosdep update
           rosdep install --from-paths src --ignore-src -y
@@ -73,6 +76,15 @@ jobs:
 
       - name: Run unit and integration tests
         timeout-minutes: 15
+        env:
+          # FastRTPS 2.6 on Humble has a known use-after-free in the
+          # discovery-teardown path (EDP::unpairWriterProxy) that segfaults
+          # peer nodes when the gateway shuts down, so we force CycloneDDS
+          # there. Rolling + Jazzy ship a newer FastRTPS without that bug
+          # and hit a separate iceoryx-shared-memory crash under CycloneDDS
+          # during shutdown ("string capacity was zero for allocated data"),
+          # so on those distros we keep the default FastRTPS.
+          RMW_IMPLEMENTATION: ${{ matrix.ros_distro == 'humble' && 'rmw_cyclonedds_cpp' || '' }}
         run: |
           source /opt/ros/${{ matrix.ros_distro }}/setup.bash
           colcon test --return-code-on-test-failure \
@@ -185,7 +197,9 @@ jobs:
       - name: Install dependencies
         run: |
           apt-get update
-          apt-get install -y ros-jazzy-test-msgs
+          apt-get install -y \
+            ros-jazzy-test-msgs \
+            ros-jazzy-rmw-cyclonedds-cpp
           source /opt/ros/jazzy/setup.bash
           rosdep update
           rosdep install --from-paths src --ignore-src -y
@@ -199,6 +213,11 @@ jobs:
         run: tar xf jazzy-build.tar && rm jazzy-build.tar
 
       - name: Run unit and integration tests
+        env:
+          # FastRTPS has a known use-after-free in discovery-teardown that
+          # can segfault peer nodes when another node (typically the gateway)
+          # shuts down. CycloneDDS avoids it.
+          RMW_IMPLEMENTATION: rmw_cyclonedds_cpp
         run: |
           source /opt/ros/jazzy/setup.bash
           colcon test --return-code-on-test-failure \

src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/rosbag_capture.hpp

@@ -165,6 +165,10 @@ class RosbagCapture {
   /// Post-fault recording state
   std::string current_fault_code_;
   std::string current_bag_path_;
+  /// Protects post_fault_timer_ against concurrent assignment in
+  /// on_fault_confirmed() (service thread) and reset in
+  /// post_fault_timer_callback() / stop() (executor thread).
+  std::mutex post_fault_timer_mutex_;
   rclcpp::TimerBase::SharedPtr post_fault_timer_;
   std::atomic<bool> recording_post_fault_{false};
 

src/ros2_medkit_fault_manager/src/rosbag_capture.cpp

@@ -135,9 +135,12 @@ void RosbagCapture::stop() {
   running_.store(false);
 
   // Cancel any pending post-fault timer
-  if (post_fault_timer_) {
-    post_fault_timer_->cancel();
-    post_fault_timer_.reset();
+  {
+    std::lock_guard<std::mutex> lock(post_fault_timer_mutex_);
+    if (post_fault_timer_) {
+      post_fault_timer_->cancel();
+      post_fault_timer_.reset();
+    }
   }
 
   if (discovery_retry_timer_) {
@@ -202,10 +205,13 @@ void RosbagCapture::on_fault_confirmed(const std::string & fault_code) {
 
     // Create timer for post-fault recording
     auto duration = std::chrono::duration<double>(config_.duration_after_sec);
-    post_fault_timer_ =
-        node_->create_wall_timer(std::chrono::duration_cast<std::chrono::nanoseconds>(duration), [this]() {
-          post_fault_timer_callback();
-        });
+    {
+      std::lock_guard<std::mutex> lock(post_fault_timer_mutex_);
+      post_fault_timer_ =
+          node_->create_wall_timer(std::chrono::duration_cast<std::chrono::nanoseconds>(duration), [this]() {
+            post_fault_timer_callback();
+          });
+    }
 
     RCLCPP_DEBUG(node_->get_logger(), "Recording %.1fs more after fault confirmation", config_.duration_after_sec);
   } else {
@@ -436,14 +442,19 @@ std::vector<std::string> RosbagCapture::resolve_topics() const {
       }
     }
   } else if (config_.topics == "all" || config_.topics == "auto") {
-    // Discover all available topics ("auto" is an alias for "all")
-    auto topic_names_and_types = node_->get_topic_names_and_types();
-    for (const auto & [topic, types] : topic_names_and_types) {
-      // Skip internal ROS topics
-      if (topic.find("/parameter_events") != std::string::npos || topic.find("/rosout") != std::string::npos) {
-        continue;
+    // Discover all available topics ("auto" is an alias for "all").
+    // Skip gracefully if the context is invalidated mid-call (shutdown race).
+    try {
+      auto topic_names_and_types = node_->get_topic_names_and_types();
+      for (const auto & [topic, types] : topic_names_and_types) {
+        // Skip internal ROS topics
+        if (topic.find("/parameter_events") != std::string::npos || topic.find("/rosout") != std::string::npos) {
+          continue;
+        }
+        topics_set.insert(topic);
       }
-      topics_set.insert(topic);
+    } catch (const std::runtime_error &) {
+      // context invalid during shutdown - no topics to add
     }
   } else if (config_.topics == "explicit") {
     // Explicit mode: use only include_topics, no topic derivation
@@ -476,10 +487,18 @@ std::vector<std::string> RosbagCapture::resolve_topics() const {
 }
 
 std::string RosbagCapture::get_topic_type(const std::string & topic) const {
-  auto topic_names_and_types = node_->get_topic_names_and_types();
-  auto it = topic_names_and_types.find(topic);
-  if (it != topic_names_and_types.end() && !it->second.empty()) {
-    return it->second[0];
+  // node_->get_topic_names_and_types() throws if the rcl context is
+  // invalidated mid-call (e.g. SIGINT fires between the check and the rcl
+  // call). During shutdown this is expected and not actionable - treat it
+  // as "no topic info available right now".
+  try {
+    auto topic_names_and_types = node_->get_topic_names_and_types();
+    auto it = topic_names_and_types.find(topic);
+    if (it != topic_names_and_types.end() && !it->second.empty()) {
+      return it->second[0];
+    }
+  } catch (const std::runtime_error &) {
+    // context invalid - caller handles empty return
   }
   return "";
 }
@@ -640,9 +659,12 @@ void RosbagCapture::post_fault_timer_callback() {
   }
 
   // Cancel timer (one-shot)
-  if (post_fault_timer_) {
-    post_fault_timer_->cancel();
-    post_fault_timer_.reset();
+  {
+    std::lock_guard<std::mutex> lock(post_fault_timer_mutex_);
+    if (post_fault_timer_) {
+      post_fault_timer_->cancel();
+      post_fault_timer_.reset();
+    }
   }
 
   // Stop post-fault recording (no more direct writes to bag)

src/ros2_medkit_fault_manager/src/snapshot_capture.cpp

@@ -348,10 +348,18 @@ std::vector<std::string> SnapshotCapture::collect_all_configured_topics() const
 }
 
 std::string SnapshotCapture::get_topic_type(const std::string & topic) const {
-  auto topic_names_and_types = node_->get_topic_names_and_types();
-  auto it = topic_names_and_types.find(topic);
-  if (it != topic_names_and_types.end() && !it->second.empty()) {
-    return it->second[0];
+  // node_->get_topic_names_and_types() throws if the rcl context is
+  // invalidated mid-call (e.g. SIGINT fires between the check and the rcl
+  // call). During shutdown this is expected and not actionable - treat it
+  // as "no topic info available right now".
+  try {
+    auto topic_names_and_types = node_->get_topic_names_and_types();
+    auto it = topic_names_and_types.find(topic);
+    if (it != topic_names_and_types.end() && !it->second.empty()) {
+      return it->second[0];
+    }
+  } catch (const std::runtime_error &) {
+    // context invalid - caller handles empty return
   }
   return "";
 }

src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp

@@ -14,10 +14,7 @@
 
 #pragma once
 
-#include <atomic>
-#include <condition_variable>
 #include <memory>
-#include <mutex>
 #include <rclcpp/rclcpp.hpp>
 #include <string>
 #include <thread>
@@ -296,9 +293,6 @@ class GatewayNode : public rclcpp::Node {
 
   // REST server thread management
   std::unique_ptr<std::thread> server_thread_;
-  std::atomic<bool> server_running_{false};
-  std::mutex server_mutex_;
-  std::condition_variable server_cv_;
 };
 
 /**

src/ros2_medkit_gateway/include/ros2_medkit_gateway/http/rest_server.hpp

@@ -81,6 +81,13 @@ class RESTServer {
     return http_server_ && http_server_->is_tls_enabled();
   }
 
+  /// True once cpp-httplib's listen() has reached its accept loop.
+  /// GatewayNode uses this as the start-up readiness signal so shutdown
+  /// cannot race a listen() that has not yet started.
+  bool is_running() const {
+    return http_server_ && http_server_->is_running();
+  }
+
  private:
   void setup_routes();
   void setup_pre_routing_handler();

src/ros2_medkit_gateway/src/gateway_node.cpp

@@ -1827,46 +1827,36 @@ void GatewayNode::refresh_cache() {
 
 void GatewayNode::start_rest_server() {
   server_thread_ = std::make_unique<std::thread>([this]() {
-    {
-      std::lock_guard<std::mutex> lock(server_mutex_);
-      server_running_ = true;
-    }
-    server_cv_.notify_all();
-
     try {
       rest_server_->start();
     } catch (const std::exception & e) {
       RCLCPP_ERROR(get_logger(), "REST server failed to start: %s", e.what());
     } catch (...) {
       RCLCPP_ERROR(get_logger(), "REST server failed to start: unknown exception");
     }
-
-    {
-      std::lock_guard<std::mutex> lock(server_mutex_);
-      server_running_ = false;
-    }
-    server_cv_.notify_all();
   });
 
-  // Wait for server to start
-  std::unique_lock<std::mutex> lock(server_mutex_);
-  server_cv_.wait(lock, [this] {
-    return server_running_.load();
-  });
+  // Wait for the server to actually reach cpp-httplib's accept loop before
+  // returning. is_running() becomes true only after listen() has passed
+  // bind and entered its select/poll loop; using a "thread started" flag
+  // here is not enough because stop() called before listen() entered its
+  // loop could be missed, leaving listen() blocking forever and the
+  // subsequent join() hanging indefinitely.
+  using namespace std::chrono_literals;
+  const auto deadline = std::chrono::steady_clock::now() + 5s;
+  while (!rest_server_->is_running() && std::chrono::steady_clock::now() < deadline) {
+    std::this_thread::sleep_for(1ms);
+  }
+  if (!rest_server_->is_running()) {
+    RCLCPP_ERROR(get_logger(), "REST server did not become ready within 5s");
+  }
 }
 
 void GatewayNode::stop_rest_server() {
   if (rest_server_) {
     rest_server_->stop();
   }
-
-  // Wait for server thread to finish
   if (server_thread_ && server_thread_->joinable()) {
-    std::unique_lock<std::mutex> lock(server_mutex_);
-    server_cv_.wait(lock, [this] {
-      return !server_running_.load();
-    });
-    lock.unlock();  // Release before join to avoid deadlock
     server_thread_->join();
   }
 }

src/ros2_medkit_integration_tests/demo_nodes/beacon_publisher.cpp

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <csignal>
 #include <diagnostic_msgs/msg/key_value.hpp>
 #include <mutex>
 #include <rclcpp/rclcpp.hpp>
@@ -84,9 +85,25 @@ class BeaconPublisher : public rclcpp::Node {
 };
 
 int main(int argc, char * argv[]) {
+  // Block SIGINT/SIGTERM until the executor has allocated its guard
+  // condition; a signal arriving mid-init invalidates the rcl context and
+  // causes rcl_* calls to throw RCLError. Unblocking after add_node() lets
+  // any queued signal be handled as a normal shutdown.
+  sigset_t mask, old;
+  sigemptyset(&mask);
+  sigaddset(&mask, SIGINT);
+  sigaddset(&mask, SIGTERM);
+  pthread_sigmask(SIG_BLOCK, &mask, &old);
+
   rclcpp::init(argc, argv);
   auto node = std::make_shared<BeaconPublisher>();
-  rclcpp::spin(node);
+  rclcpp::executors::SingleThreadedExecutor executor;
+  executor.add_node(node);
+
+  pthread_sigmask(SIG_SETMASK, &old, nullptr);
+
+  executor.spin();
+  executor.remove_node(node);
   node.reset();
   rclcpp::shutdown();
   return 0;

src/ros2_medkit_integration_tests/demo_nodes/brake_actuator.cpp

@@ -23,6 +23,7 @@
  */
 
 #include <chrono>
+#include <csignal>
 #include <mutex>
 #include <rclcpp/rclcpp.hpp>
 #include <std_msgs/msg/float32.hpp>
@@ -103,9 +104,25 @@ class BrakeActuator : public rclcpp::Node {
 };
 
 int main(int argc, char * argv[]) {
+  // Block SIGINT/SIGTERM until the executor has allocated its guard
+  // condition; a signal arriving mid-init invalidates the rcl context and
+  // causes rcl_* calls to throw RCLError. Unblocking after add_node() lets
+  // any queued signal be handled as a normal shutdown.
+  sigset_t mask, old;
+  sigemptyset(&mask);
+  sigaddset(&mask, SIGINT);
+  sigaddset(&mask, SIGTERM);
+  pthread_sigmask(SIG_BLOCK, &mask, &old);
+
   rclcpp::init(argc, argv);
   auto node = std::make_shared<BrakeActuator>();
-  rclcpp::spin(node);
+  rclcpp::executors::SingleThreadedExecutor executor;
+  executor.add_node(node);
+
+  pthread_sigmask(SIG_SETMASK, &old, nullptr);
+
+  executor.spin();
+  executor.remove_node(node);
   node.reset();
   rclcpp::shutdown();
   return 0;

src/ros2_medkit_integration_tests/demo_nodes/brake_pressure_sensor.cpp

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <csignal>
 #include <mutex>
 #include <rclcpp/rclcpp.hpp>
 #include <std_msgs/msg/float32.hpp>
@@ -60,9 +61,25 @@ class BrakePressureSensor : public rclcpp::Node {
 };
 
 int main(int argc, char ** argv) {
+  // Block SIGINT/SIGTERM until the executor has allocated its guard
+  // condition; a signal arriving mid-init invalidates the rcl context and
+  // causes rcl_* calls to throw RCLError. Unblocking after add_node() lets
+  // any queued signal be handled as a normal shutdown.
+  sigset_t mask, old;
+  sigemptyset(&mask);
+  sigaddset(&mask, SIGINT);
+  sigaddset(&mask, SIGTERM);
+  pthread_sigmask(SIG_BLOCK, &mask, &old);
+
   rclcpp::init(argc, argv);
   auto node = std::make_shared<BrakePressureSensor>();
-  rclcpp::spin(node);
+  rclcpp::executors::SingleThreadedExecutor executor;
+  executor.add_node(node);
+
+  pthread_sigmask(SIG_SETMASK, &old, nullptr);
+
+  executor.spin();
+  executor.remove_node(node);
   node.reset();
   rclcpp::shutdown();
   return 0;