diff --git a/docs/conf.py b/docs/conf.py index b144537d4..f994904fb 100644 --- a/docs/conf.py +++ b/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", ] diff --git a/src/ros2_medkit_action_status_bridge/test/test_integration.test.py b/src/ros2_medkit_action_status_bridge/test/test_integration.test.py index 28722deb9..ff4002743 100644 --- a/src/ros2_medkit_action_status_bridge/test/test_integration.test.py +++ b/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) diff --git a/src/ros2_medkit_diagnostic_bridge/test/test_integration.test.py b/src/ros2_medkit_diagnostic_bridge/test/test_integration.test.py index b0200f792..be867f732 100644 --- a/src/ros2_medkit_diagnostic_bridge/test/test_integration.test.py +++ b/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() diff --git a/src/ros2_medkit_gateway/test/test_operation_handlers.cpp b/src/ros2_medkit_gateway/test/test_operation_handlers.cpp index 65013684b..91d5241a3 100644 --- a/src/ros2_medkit_gateway/test/test_operation_handlers.cpp +++ b/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 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 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 argv; argv.reserve(args.size()); @@ -273,6 +274,12 @@ class OperationHandlersFixtureTest : public ::testing::Test { gateway_node_ = std::make_shared(); 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("test_calibrate_service", "/powertrain/engine"); trigger_service_ = service_node_->create_service( "calibrate", [](const std::shared_ptr & /*request*/, @@ -294,19 +301,20 @@ class OperationHandlersFixtureTest : public ::testing::Test { ctx_ = std::make_unique(gateway_node_.get(), cors_, auth_, tls_, nullptr); handlers_ = std::make_unique(*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; } diff --git a/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/gateway_test_case.py b/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/gateway_test_case.py index 865fec39f..f20a5837e 100644 --- a/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/gateway_test_case.py +++ b/src/ros2_medkit_integration_tests/ros2_medkit_test_utils/gateway_test_case.py @@ -845,6 +845,63 @@ def wait_for_operation(self, entity_endpoint, operation_id, *, max_wait=15.0): f'{entity_endpoint}/operations within {max_wait}s' ) + def wait_for_operation_type_info( + self, entity_endpoint, operation_id, required_keys, *, max_wait=15.0 + ): + """Wait until an operation exposes a resolved ``type_info`` schema. + + An operation appears under ``/operations`` as soon as it is discovered + by name, but its ``x-medkit.type_info`` is only populated once the + gateway has resolved the underlying ROS interface type: while + ``ros2.type`` is still empty, or the type-introspection lookup has not + yet loaded the type, the handler omits ``type_info`` entirely. Polling + the operation-name endpoint alone (``wait_for_operation``) races that + second resolution step, so wait until ``type_info`` actually carries the + expected schema keys (``goal``/``result``/``feedback`` for an action, + ``request``/``response`` for a service) before asserting on it. + + Parameters + ---------- + entity_endpoint : str + Entity path relative to ``BASE_URL`` (e.g. ``'/apps/long_calibration'``). + operation_id : str + The operation ID (matched against each item's ``id`` or ``name``). + required_keys : iterable of str + Schema keys that must all be present in ``type_info``. + max_wait : float + Maximum time to wait in seconds. + + Returns + ------- + dict + The operation item whose ``type_info`` carries all *required_keys*. + + Raises + ------ + AssertionError + If no such operation is found within *max_wait*. + + """ + required = tuple(required_keys) + + def _resolved(data): + for op in data.get('items', []): + if op.get('id') != operation_id and op.get('name') != operation_id: + continue + type_info = op.get('x-medkit', {}).get('type_info') + if isinstance(type_info, dict) and all( + key in type_info for key in required + ): + return op + return None + + return self.poll_endpoint_until( + f'{entity_endpoint}/operations', + _resolved, + timeout=max_wait, + interval=0.5, + ) + def wait_for_data_item(self, entity_endpoint, direction, *, timeout=10.0): """Poll entity data until an item with the given direction appears. diff --git a/src/ros2_medkit_integration_tests/test/features/test_operations_api.test.py b/src/ros2_medkit_integration_tests/test/features/test_operations_api.test.py index 1238a9d93..c63676e61 100644 --- a/src/ros2_medkit_integration_tests/test/features/test_operations_api.test.py +++ b/src/ros2_medkit_integration_tests/test/features/test_operations_api.test.py @@ -259,21 +259,17 @@ def test_root_endpoint_includes_operations(self): def test_service_operation_has_type_info_schema(self): """Service operations include type_info with request/response schemas.""" - ops_data = self.get_json('/apps/calibration/operations') - self.assertIn('items', ops_data, 'Operations endpoint should return items') - ops = ops_data['items'] - - # Find the calibrate service operation - calibrate_op = None - for op in ops: - if op['name'] == 'calibrate': - x_medkit = op.get('x-medkit', {}) - ros2 = x_medkit.get('ros2', {}) - if ros2.get('kind') == 'service': - calibrate_op = op - break + # The operation is listed by name before its service type is resolved, + # so wait until type_info carries the request/response schema keys + # rather than reading the list once and racing type resolution. + calibrate_op = self.wait_for_operation_type_info( + '/apps/calibration', 'calibrate', ('request', 'response'), + ) - self.assertIsNotNone(calibrate_op, 'Calibrate service should be listed') + # Confirm it is the service operation. + ros2 = calibrate_op.get('x-medkit', {}).get('ros2', {}) + self.assertEqual(ros2.get('kind'), 'service', + 'Calibrate service should be listed') # Verify type_info is present in x-medkit with request/response schemas x_medkit = calibrate_op['x-medkit'] diff --git a/src/ros2_medkit_integration_tests/test/scenarios/test_scenario_action_lifecycle.test.py b/src/ros2_medkit_integration_tests/test/scenarios/test_scenario_action_lifecycle.test.py index 3fefa6971..2846f6077 100644 --- a/src/ros2_medkit_integration_tests/test/scenarios/test_scenario_action_lifecycle.test.py +++ b/src/ros2_medkit_integration_tests/test/scenarios/test_scenario_action_lifecycle.test.py @@ -216,20 +216,18 @@ def test_05_action_has_type_info_schema(self): """GET operation details, verify type_info with goal/result/feedback schemas.""" self._ensure_operation_ready() - ops_data = self.get_json(f'{self.ENTITY_ENDPOINT}/operations') - ops = ops_data['items'] - - # Find the long_calibration action operation - action_op = None - for op in ops: - if op['name'] == self.OPERATION_ID: - x_medkit = op.get('x-medkit', {}) - ros2 = x_medkit.get('ros2', {}) - if ros2.get('kind') == 'action': - action_op = op - break + # The operation appears by name before its interface type is resolved, + # so wait until type_info actually carries the action schema keys rather + # than reading the list once and racing type resolution. + action_op = self.wait_for_operation_type_info( + self.ENTITY_ENDPOINT, self.OPERATION_ID, + ('goal', 'result', 'feedback'), + ) - self.assertIsNotNone(action_op, 'Long calibration action should be listed') + # Confirm it is the action operation (not a same-named service). + ros2 = action_op.get('x-medkit', {}).get('ros2', {}) + self.assertEqual(ros2.get('kind'), 'action', + 'Long calibration action should be listed') # Verify type_info in x-medkit x_medkit = action_op['x-medkit'] diff --git a/src/ros2_medkit_plugins/ros2_medkit_opcua/include/ros2_medkit_opcua/opcua_plugin.hpp b/src/ros2_medkit_plugins/ros2_medkit_opcua/include/ros2_medkit_opcua/opcua_plugin.hpp index 8e4562ca8..3ed0587d1 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_opcua/include/ros2_medkit_opcua/opcua_plugin.hpp +++ b/src/ros2_medkit_plugins/ros2_medkit_opcua/include/ros2_medkit_opcua/opcua_plugin.hpp @@ -30,6 +30,7 @@ #include #include +#include #include #include #include @@ -136,6 +137,12 @@ class OpcuaPlugin : public ros2_medkit_gateway::GatewayPlugin, const std::string & severity_str, const std::string & message); void send_clear_fault(const std::string & fault_code); + // Dispatch now if the fault_manager service is matched, else buffer the + // dispatch (bounded, order-preserving) to be flushed once it appears. + void send_or_buffer(std::function dispatch); + // Flush buffered fault dispatches when the fault_manager service is ready. + void flush_pending_reports(); + // Publish PLC values to ROS 2 topics (called after each poll) void publish_values(const PollSnapshot & snap); @@ -168,6 +175,14 @@ class OpcuaPlugin : public ros2_medkit_gateway::GatewayPlugin, struct FaultClients; std::unique_ptr fault_clients_; + // Ordered buffer of pending fault report/clear dispatches. ReportFault / + // ClearFault are fire-and-forget, so a report sent before the fault_manager + // service is DDS-matched is dropped. Instead of blocking startup for the sink, + // buffer the dispatch (preserving report-then-clear order) and flush it on the + // next poll once the service is ready. Poll-thread only (send_report_fault / + // send_clear_fault via on_alarm_change, flush via the poll callback), no lock. + std::vector> pending_reports_; + // Tracks which non-numeric nodes have already been warned about (avoids log spam). // Instance member instead of static to survive plugin reload (dlclose/dlopen). std::unordered_set warned_non_numeric_; diff --git a/src/ros2_medkit_plugins/ros2_medkit_opcua/src/opcua_plugin.cpp b/src/ros2_medkit_plugins/ros2_medkit_opcua/src/opcua_plugin.cpp index 2c8fd7a91..f00977e6d 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_opcua/src/opcua_plugin.cpp +++ b/src/ros2_medkit_plugins/ros2_medkit_opcua/src/opcua_plugin.cpp @@ -359,6 +359,14 @@ void OpcuaPlugin::set_context(PluginContext & context) { poller_config_.log_warn = [this](const std::string & msg) { log_warn(msg); }; + + // Do not block startup on the fault sink. OPC-UA AlarmCondition notifications + // are one-shot (reported once on the inactive->active transition) and + // ReportFault is fire-and-forget, so a report raised before fault_manager is + // DDS-matched would be lost. send_report_fault/send_clear_fault buffer the + // dispatch and flush_pending_reports (run on every poll) drains it once the + // service appears, so a late sink still receives the alarm - without stalling + // startup or capping recovery at a fixed timeout. poller_->start(poller_config_); log_info("OPC-UA poller started (mode: " + std::string(poller_->using_subscriptions() ? "subscription" : "poll") + ")"); @@ -785,7 +793,9 @@ void OpcuaPlugin::send_report_fault(const std::string & entity_id, const std::st request->severity = ros2_medkit_msgs::msg::Fault::SEVERITY_INFO; } - fault_clients_->report->async_send_request(request); + send_or_buffer([this, request]() { + fault_clients_->report->async_send_request(request); + }); } void OpcuaPlugin::send_clear_fault(const std::string & fault_code) { @@ -797,13 +807,41 @@ void OpcuaPlugin::send_clear_fault(const std::string & fault_code) { auto request = std::make_shared(); request->fault_code = fault_code; - fault_clients_->clear->async_send_request(request); + send_or_buffer([this, request]() { + fault_clients_->clear->async_send_request(request); + }); +} + +void OpcuaPlugin::send_or_buffer(std::function dispatch) { + // Bound the buffer so a deployment with no fault_manager cannot grow it + // without limit; drop the oldest (least relevant) pending dispatch. + constexpr size_t kMaxPendingReports = 256; + if (pending_reports_.size() >= kMaxPendingReports) { + pending_reports_.erase(pending_reports_.begin()); + log_warn("pending fault report buffer full (" + std::to_string(kMaxPendingReports) + "), dropping oldest"); + } + pending_reports_.push_back(std::move(dispatch)); + // Drains immediately (in order) if the sink is already matched. + flush_pending_reports(); +} + +void OpcuaPlugin::flush_pending_reports() { + if (pending_reports_.empty() || !fault_clients_->report || !fault_clients_->report->service_is_ready()) { + return; + } + for (auto & dispatch : pending_reports_) { + dispatch(); + } + pending_reports_.clear(); } void OpcuaPlugin::publish_values(const PollSnapshot & snap) { if (shutdown_requested_.load()) { return; } + // Poll-thread hook: drain any fault reports buffered before fault_manager was + // discovered, so a late sink still receives them. + flush_pending_reports(); for (const auto & [node_id, value] : snap.values) { auto pub_it = publishers_.find(node_id); if (pub_it == publishers_.end()) { diff --git a/tsan_suppressions.txt b/tsan_suppressions.txt index 236485a95..1d66c38be 100644 --- a/tsan_suppressions.txt +++ b/tsan_suppressions.txt @@ -98,3 +98,34 @@ called_from_lib:libyaml-cpp.so.0.8 race:dynmsg::cpp::impl::basic_value_to_yaml* race:dynmsg::cpp::impl::member_to_yaml* race:dynmsg::get_vector_size* +# Same staging race on the response buffer's own lifetime rather than its +# contents: the executor thread allocates the shared response inside +# rclcpp::GenericClient::create_response(), and the caller thread releases +# the last shared_ptr reference (control-block refcount decrement and the +# subsequent operator delete) after future::get() returns. libstdc++ +# future synchronisation is invisible to TSan (see above), so it reports a +# race between create_response()'s new and the caller's ~shared_ptr on the +# same block. Match the rclcpp allocation frame, which is on every one of +# these reports, so a genuine refcount race in our own shared_ptr use is +# still surfaced. +race:rclcpp::GenericClient::create_response* +# Same benign GenericClient response-staging race as create_response above, +# matched on our own frame instead of the rclcpp one. In test_operation_handlers +# under load, TSan reports the executor thread staging the response +# (rclcpp::GenericClient::create_response + the response std::string copies) +# racing the caller thread that consumes/frees it after future::get(). All the +# racing WRITES live in rclcpp/libstdc++ (create_response, std::string mutate); +# our OperationManager::call_service frame is only the consumer of a value the +# std::future contract hands us once the promise is satisfied - the +# happens-before that rclcpp establishes via pthread_once is invisible to TSan. +# We must match on this frame because the adjacent librclcpp create_response +# frame is frequently UNSYMBOLIZED ( at librclcpp.so+0x14xxxx), so the +# create_response symbol line above does not catch those reports. A library +# scope (called_from_lib:librclcpp.so) does NOT catch them either: TSan reports +# this race with our call_service stack as the primary, where librclcpp is not a +# caller, so the library filter misses it and the race resurfaces - verified, it +# fails sanitizer-tsan with the child exiting 66. Matching our own frame is the +# only reliable option: a deliberate, narrow exception to the "never suppress +# our code" rule, scoped to the service-call path only. The race itself is +# entirely a library one; we are on the stack only as the caller of the client. +race:ros2_medkit_gateway::OperationManager::call_service*