ci: harden flaky suite checks (opcua A&C, action-status, diagnostic_bridge, operation_handlers, type_info, docs linkcheck)

docs/conf.py

@@ -163,4 +163,8 @@
     r"https://github\.com/selfpatch/ros2_medkit/blob/.+#L\d+",
     # GitHub issue links may 404 in linkcheck without auth token
     r"https://github\.com/selfpatch/ros2_medkit/issues/\d+",
+    # The rep-0149 page intermittently exceeds the read timeout in CI; the
+    # canonical REP host is slow, not broken. Scope the skip to this exact URL
+    # so other REP links stay checked.
+    r"https://ros\.org/reps/rep-0149\.html",
 ]

src/ros2_medkit_action_status_bridge/test/test_integration.test.py

@@ -128,7 +128,8 @@ def list_faults(self, statuses=None):
         request.statuses = statuses or []
         future = self.list_faults_client.call_async(request)
         rclpy.spin_until_future_complete(self.node, future, timeout_sec=5.0)
-        return future.result().faults
+        result = future.result()
+        return result.faults if result is not None else []
 
     def find(self, faults, code):
         return next((f for f in faults if f.fault_code == code), None)
@@ -144,8 +145,34 @@ def wait_for_fault(self, code, statuses=None, want_status=None, timeout=15.0):
             time.sleep(0.5)
         return fault
 
+    def wait_for_bridge_watching(self, timeout=15.0):
+        """
+        Wait until the bridge subscribes to the status topic with a resolved name.
+
+        The bridge fixes a fault's source at first report: the status publisher's
+        node FQN if DDS discovery has resolved it, else the action-name fallback
+        (reporter_for). If the ABORTED status is published before the bridge has
+        resolved this node's name, the source freezes to the fallback and is never
+        re-attributed. Seeing the bridge's subscription here with a resolved node
+        name means discovery between the two participants has settled both ways, so
+        the first report resolves the server FQN instead of racing node-name
+        propagation.
+        """
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            infos = self.node.get_subscriptions_info_by_topic(STATUS_TOPIC)
+            if any(info.node_name == 'action_status_bridge' for info in infos):
+                return True
+            rclpy.spin_once(self.node, timeout_sec=0.1)
+        return False
+
     def test_01_aborted_goal_raises_fault(self):
         """An ABORTED goal on a discovered status topic raises a fault."""
+        # Synchronize on the bridge's subscription before publishing so the server
+        # FQN is resolved at first report rather than racing DDS node-name
+        # propagation (which freezes the source to the action-name fallback).
+        self.assertTrue(self.wait_for_bridge_watching(),
+                        'bridge did not subscribe to the status topic in time')
         self.status_pub.publish(_status_array(GoalStatus.STATUS_ABORTED, 1))
         # Poll through runtime discovery (rescan) + subscription + processing.
         fault = self.wait_for_fault(ABORTED_CODE)

src/ros2_medkit_diagnostic_bridge/test/test_integration.test.py

@@ -89,8 +89,30 @@ def setUpClass(cls):
         assert cls.list_faults_client.wait_for_service(timeout_sec=10.0), \
             'ListFaults service not available'
 
-        # Wait for bridge to connect
-        time.sleep(1.0)
+        # Wait for the bridge's /diagnostics subscription to match this
+        # publisher before any test publishes. The bridge subscribes with
+        # volatile QoS, so a sample published before the match is silently
+        # dropped and never becomes a fault - a fixed sleep here raced that
+        # match under load.
+        assert cls._wait_for_bridge_subscription(timeout_sec=15.0), \
+            'diagnostic_bridge did not subscribe to /diagnostics'
+
+    @classmethod
+    def _wait_for_bridge_subscription(cls, *, timeout_sec=15.0):
+        """
+        Wait until the bridge's /diagnostics subscription matches our publisher.
+
+        Polls the publisher's matched-subscription count (spinning so DDS
+        discovery can progress) instead of sleeping a fixed interval, so
+        published diagnostics are guaranteed to be delivered rather than
+        dropped by the volatile QoS during the discovery window.
+        """
+        deadline = time.monotonic() + timeout_sec
+        while time.monotonic() < deadline:
+            if cls.diag_pub.get_subscription_count() >= 1:
+                return True
+            rclpy.spin_once(cls.node, timeout_sec=0.1)
+        return False
 
     @classmethod
     def tearDownClass(cls):
@@ -99,14 +121,15 @@ def tearDownClass(cls):
         rclpy.shutdown()
 
     def list_faults(self, statuses=None):
-        """Get faults from FaultManager."""
+        """Get faults from FaultManager (empty list if the call times out)."""
         request = ListFaults.Request()
         request.filter_by_severity = False
         request.statuses = statuses or []
 
         future = self.list_faults_client.call_async(request)
         rclpy.spin_until_future_complete(self.node, future, timeout_sec=5.0)
-        return future.result().faults
+        result = future.result()
+        return result.faults if result is not None else []
 
     def publish_diagnostic(self, name, level, message='Test message'):
         """Publish a single diagnostic message."""
@@ -121,65 +144,93 @@ def publish_diagnostic(self, name, level, message='Test message'):
         self.diag_pub.publish(msg)
 
         # Give time for message to be processed
-        time.sleep(0.5)
+        time.sleep(0.3)
+
+    def publish_until(self, name, level, expected_code, *, predicate=None,
+                      message='Test message', statuses=None, timeout=25.0):
+        """
+        Republish a diagnostic until a matching fault satisfies *predicate*.
+
+        Two discovery hops gate the first fault: the bridge's /diagnostics
+        subscription must match this publisher (handled in setUpClass) and the
+        bridge's ReportFault client must have discovered fault_manager, else
+        ``send_report`` drops the fire-and-forget request. That client readiness
+        is not observable from here and the report is one-shot, so re-emit the
+        stimulus and poll rather than publishing once and racing the hop.
+        Re-emission also drives WARN/PASSED past their occurrence threshold
+        within the filter window deterministically.
+
+        Returns the matching fault once *predicate* holds; fails at the
+        deadline.
+        """
+        if predicate is None:
+            def predicate(_fault):
+                return True
+        deadline = time.monotonic() + timeout
+        last = None
+        while time.monotonic() < deadline:
+            self.publish_diagnostic(name, level, message)
+            fault = next(
+                (f for f in self.list_faults(statuses=statuses)
+                 if f.fault_code == expected_code),
+                None,
+            )
+            if fault is not None:
+                last = fault
+                if predicate(fault):
+                    return fault
+        self.fail(
+            f'Fault {expected_code} not satisfied within {timeout}s '
+            f'(last seen: {last})'
+        )
 
     def test_01_error_diagnostic_creates_fault(self):
         """Test that ERROR diagnostic creates a fault in FaultManager."""
-        self.publish_diagnostic('test_sensor', DiagnosticStatus.ERROR, 'Sensor failure')
-
-        faults = self.list_faults()
-        fault_codes = [f.fault_code for f in faults]
-
-        self.assertIn('TEST_SENSOR', fault_codes)
-        fault = next(f for f in faults if f.fault_code == 'TEST_SENSOR')
+        fault = self.publish_until(
+            'test_sensor', DiagnosticStatus.ERROR, 'TEST_SENSOR',
+            message='Sensor failure',
+        )
         self.assertEqual(fault.severity, Fault.SEVERITY_ERROR)
 
     def test_02_warn_diagnostic_creates_fault(self):
-        """Test that WARN diagnostic creates a fault."""
-        # Send multiple times to bypass FaultReporter's local filtering (default threshold=3)
-        for _ in range(3):
-            self.publish_diagnostic('warning_component', DiagnosticStatus.WARN, 'Low battery')
-
-        faults = self.list_faults()
-        fault_codes = [f.fault_code for f in faults]
-
-        self.assertIn('WARNING_COMPONENT', fault_codes)
-        fault = next(f for f in faults if f.fault_code == 'WARNING_COMPONENT')
+        """
+        Test that WARN diagnostic creates a fault.
+
+        WARN does not bypass the reporter's local filter (threshold=3), so
+        publish_until re-emits until the occurrence threshold is met and the
+        report reaches fault_manager.
+        """
+        fault = self.publish_until(
+            'warning_component', DiagnosticStatus.WARN, 'WARNING_COMPONENT',
+            message='Low battery',
+        )
         self.assertEqual(fault.severity, Fault.SEVERITY_WARN)
 
     def test_03_stale_diagnostic_creates_critical_fault(self):
         """Test that STALE diagnostic creates a CRITICAL fault."""
-        self.publish_diagnostic('stale_sensor', DiagnosticStatus.STALE, 'No data')
-
-        faults = self.list_faults()
-        fault_codes = [f.fault_code for f in faults]
-
-        self.assertIn('STALE_SENSOR', fault_codes)
-        fault = next(f for f in faults if f.fault_code == 'STALE_SENSOR')
+        fault = self.publish_until(
+            'stale_sensor', DiagnosticStatus.STALE, 'STALE_SENSOR',
+            message='No data',
+        )
         self.assertEqual(fault.severity, Fault.SEVERITY_CRITICAL)
 
     def test_04_ok_diagnostic_heals_fault(self):
         """Test that OK diagnostic sends PASSED and heals fault."""
-        # First create a fault
-        self.publish_diagnostic('healing_test', DiagnosticStatus.ERROR, 'Error')
-
-        faults = self.list_faults()
-        self.assertIn('HEALING_TEST', [f.fault_code for f in faults])
-
-        # Send OK multiple times to bypass FaultReporter's local PASSED filtering
-        # (default threshold=3, same as FAILED events). Extra iterations ensure
-        # the async service call has time to reach FaultManager.
-        for _ in range(4):
-            self.publish_diagnostic('healing_test', DiagnosticStatus.OK)
-
-        # Allow time for the async PASSED service call to be processed
-        time.sleep(1.0)
-
-        # Check fault is healed (query all statuses to find it)
-        faults = self.list_faults(statuses=[Fault.STATUS_CONFIRMED, Fault.STATUS_HEALED])
-        fault = next((f for f in faults if f.fault_code == 'HEALING_TEST'), None)
-        self.assertIsNotNone(fault, 'HEALING_TEST fault not found')
-        # After PASSED events with healing_threshold=1, fault should be HEALED
+        # First create a fault.
+        self.publish_until(
+            'healing_test', DiagnosticStatus.ERROR, 'HEALING_TEST',
+            message='Error',
+        )
+
+        # Send OK until the fault heals. PASSED events share the reporter's
+        # local threshold (3) and the healing service call is async, so
+        # re-emit and poll until the fault reaches HEALED instead of publishing
+        # a fixed number of times and racing propagation.
+        fault = self.publish_until(
+            'healing_test', DiagnosticStatus.OK, 'HEALING_TEST',
+            predicate=lambda f: f.status == Fault.STATUS_HEALED,
+            statuses=[Fault.STATUS_CONFIRMED, Fault.STATUS_HEALED],
+        )
         self.assertEqual(fault.status, Fault.STATUS_HEALED)
 
     def test_05_fault_code_generation(self):
@@ -191,13 +242,9 @@ def test_05_fault_code_generation(self):
         ]
 
         for diag_name, expected_code in test_cases:
-            self.publish_diagnostic(diag_name, DiagnosticStatus.ERROR)
-
-            faults = self.list_faults()
-            fault_codes = [f.fault_code for f in faults]
-
-            self.assertIn(expected_code, fault_codes,
-                          f'Expected {expected_code} from diagnostic "{diag_name}"')
+            self.publish_until(
+                diag_name, DiagnosticStatus.ERROR, expected_code,
+            )
 
 
 @launch_testing.post_shutdown_test()

src/ros2_medkit_gateway/test/test_operation_handlers.cpp

@@ -245,14 +245,15 @@ class OperationHandlersFixtureTest : public ::testing::Test {
     suite_server_port_ = reserve_local_port();
     ASSERT_NE(suite_server_port_, 0);
 
-    std::vector<std::string> args = {"test_operation_handlers",
-                                     "--ros-args",
-                                     "-p",
-                                     "server.port:=" + std::to_string(suite_server_port_),
-                                     "-p",
-                                     "refresh_interval_ms:=60000",
-                                     "-p",
-                                     "service_call_timeout_sec:=1"};
+    std::vector<std::string> args = {"test_operation_handlers", "--ros-args", "-p",
+                                     "server.port:=" + std::to_string(suite_server_port_), "-p",
+                                     "refresh_interval_ms:=60000", "-p",
+                                     // Pin the graph-event refresh debounce at its 60s maximum so the
+                                     // single startup refresh_cache() that reconciles the test nodes is
+                                     // the only one for the test's lifetime. Without this, a refresh
+                                     // fired by the goal's client/subscription graph churn could wipe the
+                                     // manually seeded component between the seed and list_executions.
+                                     "discovery.refresh_debounce_ms:=60000", "-p", "service_call_timeout_sec:=1"};
 
     std::vector<char *> argv;
     argv.reserve(args.size());
@@ -273,6 +274,12 @@ class OperationHandlersFixtureTest : public ::testing::Test {
     gateway_node_ = std::make_shared<GatewayNode>();
     ASSERT_NE(gateway_node_, nullptr);
 
+    // Cache generation right after construction: only the gateway's own graph is
+    // discovered at this point. The first graph-event-driven refresh_cache()
+    // after the test nodes join advances it, which is the signal we wait on
+    // before seeding.
+    const uint64_t base_generation = gateway_node_->get_thread_safe_cache().generation();
+
     service_node_ = std::make_shared<rclcpp::Node>("test_calibrate_service", "/powertrain/engine");
     trigger_service_ = service_node_->create_service<std_srvs::srv::Trigger>(
         "calibrate", [](const std::shared_ptr<std_srvs::srv::Trigger::Request> & /*request*/,
@@ -294,19 +301,20 @@ class OperationHandlersFixtureTest : public ::testing::Test {
     ctx_ = std::make_unique<HandlerContext>(gateway_node_.get(), cors_, auth_, tls_, nullptr);
     handlers_ = std::make_unique<OperationHandlers>(*ctx_);
 
-    // Wait for DDS discovery to complete: the executor must discover
-    // the action server, trigger service, and all internal action services
-    // (_cancel_goal, _get_result, _status). On slow CI runners under
-    // parallel load, 200ms was insufficient.
-    std::this_thread::sleep_for(1s);
-
-    // Seed the component cache AFTER discovery has settled. The gateway's
-    // graph-event-driven `refresh_cache()` fires whenever the executor
-    // observes a graph change (adding service_node_ / action_server_node_
-    // produces several such events), and each refresh pass calls
-    // `cache.update_all(...)` from the gateway's own discovery view -
-    // which would otherwise wipe a pre-spin seed. Seeding here guarantees
-    // the test's manually-injected entities are the latest write.
+    // Deterministic readiness, replacing a fixed discovery sleep. Two signals
+    // must hold before we seed and drive operations:
+    //   1. The action's send_goal service and the trigger service are visible
+    //      on the gateway participant, so create_execution's send_goal
+    //      (wait_for_service) resolves instead of racing DDS discovery.
+    //   2. The cache generation advanced past construction, proving the single
+    //      graph-event-driven refresh_cache() that reconciles these nodes has
+    //      already run. With discovery.refresh_debounce_ms pinned at 60s, that
+    //      first refresh is the only one for the test's lifetime, so seeding
+    //      afterwards is the final cache write and cannot be clobbered by a
+    //      concurrent refresh mid-test (the race that made ListExecutions flake
+    //      under load).
+    ASSERT_TRUE(wait_for_discovery_settled(base_generation))
+        << "action/service discovery did not settle before seeding";
     seed_component_cache();
   }
 
@@ -334,6 +342,33 @@ class OperationHandlersFixtureTest : public ::testing::Test {
     gateway_node_.reset();
   }
 
+  /// Block until DDS discovery has surfaced the action's send_goal service and
+  /// the trigger service on the gateway participant AND the gateway has run its
+  /// first post-startup refresh_cache() (cache generation advanced past
+  /// `base_generation`). Returns false if either signal is missing at the
+  /// deadline. This is a real readiness gate, not a fixed sleep.
+  bool wait_for_discovery_settled(uint64_t base_generation, std::chrono::seconds timeout = std::chrono::seconds(15)) {
+    const std::string send_goal_srv = "/powertrain/engine/long_calibration/_action/send_goal";
+    const std::string calibrate_srv = "/powertrain/engine/calibrate";
+    const auto deadline = std::chrono::steady_clock::now() + timeout;
+    bool graph_ready = false;
+    bool refresh_ran = false;
+    while (std::chrono::steady_clock::now() < deadline) {
+      if (!graph_ready) {
+        const auto services = gateway_node_->get_service_names_and_types();
+        graph_ready = services.count(send_goal_srv) > 0 && services.count(calibrate_srv) > 0;
+      }
+      if (!refresh_ran) {
+        refresh_ran = gateway_node_->get_thread_safe_cache().generation() > base_generation;
+      }
+      if (graph_ready && refresh_ran) {
+        return true;
+      }
+      std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    }
+    return false;
+  }
+
   void seed_component_cache() {
     Component component;
     component.id = "engine";
@@ -373,8 +408,9 @@ class OperationHandlersFixtureTest : public ::testing::Test {
     }
     EXPECT_FALSE(async_ptr->id.empty());
 
-    // Re-seed cache after the goal subscription perturbs discovery.
-    seed_component_cache();
+    // No re-seed needed: with the refresh debounce pinned at 60s, the goal's
+    // client/subscription graph churn cannot trigger a refresh that wipes the
+    // seed within the test's lifetime.
     return async_ptr->id;
   }