From 863f34945b270a8e911a37b7d2c555282256e6d9 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Wed, 12 Aug 2026 17:18:56 +0200 Subject: [PATCH 1/4] feat(fault_manager)!: write black-box bags as mcap by default and declare the storage plugins The package declared no rosbag2 storage plugin at all. `rosbag2_storage` is the plugin interface and `libsqlite3-dev` is the C library the fault database uses; neither provides a backend. An installed fault manager therefore found no storage, logged a warning and disabled black-box capture, which the README leads with. The Humble binarydeb job is the only place these packages meet a minimal dependency closure, and it is where this surfaced. The default format returns to mcap, which is what rosbag2 has defaulted to since Iron and what Foxglove and Lichtblick open without conversion. This restores the default the project chose at 0.4.0; it was moved to sqlite3 in a later fix on the stated premise that sqlite3 always ships, and that premise was false for the same reason as above. Both plugins are now declared as exec and test dependencies, and the runtime stage of the image installs them, so the published image can write the format it advertises. Storage resolution is symmetric. Whichever format is configured, if its plugin fails to load the other one is probed and used, with a warning naming the missing package; capture disables only when neither loads, naming both packages. BREAKING CHANGE: an installation that upgrades without setting `snapshots.rosbag.format` starts producing `.mcap` bags instead of `.db3`, and the bulk-data download filename changes with it. Anything reading the old extension directly has to follow. --- Dockerfile | 24 ++ QUALITY_DECLARATION.md | 2 + docs/config/fault-manager.rst | 12 +- docs/tutorials/snapshots.rst | 37 +- src/ros2_medkit_fault_manager/CMakeLists.txt | 13 + src/ros2_medkit_fault_manager/README.md | 17 + .../config/snapshots.yaml | 17 +- .../snapshot_capture.hpp | 10 +- src/ros2_medkit_fault_manager/package.xml | 17 +- .../src/fault_manager_node.cpp | 2 +- .../src/rosbag_capture.cpp | 50 +-- .../test/test_rosbag_capture.cpp | 54 ++- .../test/test_rosbag_default_format.test.py | 362 ++++++++++++++++++ .../test/test_rosbag_storage_dependency.py | 123 ++++++ src/ros2_medkit_gateway/README.md | 2 +- .../src/http/handlers/bulkdata_handlers.cpp | 8 +- 16 files changed, 681 insertions(+), 69 deletions(-) create mode 100644 src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py create mode 100644 src/ros2_medkit_fault_manager/test/test_rosbag_storage_dependency.py diff --git a/Dockerfile b/Dockerfile index 86852f57f..805cde578 100644 --- a/Dockerfile +++ b/Dockerfile @@ -80,6 +80,17 @@ COPY src/ros2_medkit_plugins/ ${COLCON_WS}/src/ros2_medkit_plugins/ # test_depend and which rosdep tries to install even with BUILD_TESTING=OFF). # This was previously masked by Docker layer cache hits on CI; cold builds # always failed. +# +# rosbag2_storage_mcap stays skip-keyed here even though fault_manager now +# exec_depends on it: this stage compiles with -DBUILD_TESTING=OFF against +# rosbag2_cpp/rosbag2_storage (the plugin interface, already resolvable +# without it) and never runs a bag, so it never needs the plugin itself, only +# its interface. The key exists because the plugin package was not available +# via rosdep on every distro this Dockerfile targets when it was added +# (ce702f61); that constraint is about apt package availability per distro, +# not about whether fault_manager declares the dependency, so becoming an +# exec_depend does not remove the reason to skip it here. It is installed for +# real in the runtime stage below, which is where it is actually loaded. RUN bash -c "source /opt/ros/${ROS_DISTRO}/setup.bash && \ apt-get update && \ rosdep update && \ @@ -104,10 +115,23 @@ ENV COLCON_WS=/home/medkit/ws # CycloneDDS RMW is bundled so the gateway can attach to a stack running # CycloneDDS (e.g. Autoware) without rebuilding. FastDDS stays the default; # switch with RMW_IMPLEMENTATION=rmw_cyclonedds_cpp at runtime. +# rosbag2-storage-mcap and rosbag2-storage-default-plugins are the actual +# rosbag2 storage plugins fault_manager loads at runtime through +# rosbag2_storage's plugin interface: the former is always the mcap plugin, +# the latter is the sqlite3 plugin itself on humble but, on jazzy (this +# stage's default) and lyrical, a metapackage that pulls in both sqlite3 and +# mcap - so it is not a clean sqlite3-only counterpart to the first package on +# every distro this image targets, just the name that guarantees sqlite3 is +# present. libsqlite3-0 alone is the C library, not a rosbag2 plugin, and this +# stage never gets /opt/ros from the builder - only the colcon workspace +# install/ - so without these two the image would advertise mcap-by-default +# and silently have no working black-box storage backend at all. RUN apt-get update && apt-get install -y --no-install-recommends \ ros-${ROS_DISTRO}-yaml-cpp-vendor \ ros-${ROS_DISTRO}-example-interfaces \ ros-${ROS_DISTRO}-rmw-cyclonedds-cpp \ + ros-${ROS_DISTRO}-rosbag2-storage-mcap \ + ros-${ROS_DISTRO}-rosbag2-storage-default-plugins \ libsqlite3-0 \ libsystemd0 \ libssl3 \ diff --git a/QUALITY_DECLARATION.md b/QUALITY_DECLARATION.md index 2df2a06ca..f05ef325a 100644 --- a/QUALITY_DECLARATION.md +++ b/QUALITY_DECLARATION.md @@ -195,6 +195,8 @@ Linter tests are enforced in CI on every pull request. | `diagnostic_msgs` | Level 1 | | `rosbag2_cpp` | Level 3 | | `rosbag2_storage` | Level 3 | +| `rosbag2_storage_mcap` | Level 3 | +| `rosbag2_storage_default_plugins` | Level 3 | ### Optional Direct Runtime Non-ROS Dependencies [5.ii] diff --git a/docs/config/fault-manager.rst b/docs/config/fault-manager.rst index e837bf481..6ef4991e4 100644 --- a/docs/config/fault-manager.rst +++ b/docs/config/fault-manager.rst @@ -244,7 +244,7 @@ Capture continuous rosbag recordings around fault events. exclude_topics: [] # Topics to exclude exclude_sensor_topics: true # Auto-exclude image/points/depth/compressed in broad modes lazy_start: false # Start recording on first fault - format: "sqlite3" # Storage format + format: "mcap" # Storage format (default: mcap) qos_match: true # Match each topic's publisher QoS storage_path: "" # Custom storage path max_buffer_mb: 256 # Ring-buffer RAM cap @@ -306,6 +306,14 @@ Capture continuous rosbag recordings around fault events. - ``true`` - Subscribe with each topic's publisher-offered QoS for faithful capture instead of forcing best-effort. + * - ``rosbag.format`` + - ``mcap`` + - Bag storage format: ``mcap`` (default; opens directly in Foxglove and + Lichtblick) or ``sqlite3`` (also what an unknown format string lands + on). Neither is privileged: whichever is configured, an unavailable + plugin falls back automatically to the other one, and capture disables + itself only if neither loads. Both plugins are runtime dependencies of + ``ros2_medkit_fault_manager``. * - ``rosbag.max_buffer_mb`` - ``256`` - Ring-buffer RAM cap; oldest buffered messages drop past it. @@ -384,7 +392,7 @@ both stay - the sweep is about the whole store, not about this recording, and it runs again on the next capture. A post-fault-only bag on a quiet or heavily filtered system can contain zero -messages. It still finalises normally on both ``sqlite3`` and ``mcap``, is listed +messages. It still finalises normally on both ``mcap`` and ``sqlite3``, is listed by the bulk-data endpoints and can be downloaded; only its payload is empty. **What ``duration_sec`` on a stored bag means.** The value returned by diff --git a/docs/tutorials/snapshots.rst b/docs/tutorials/snapshots.rst index 1cf038099..6d8dd5134 100644 --- a/docs/tutorials/snapshots.rst +++ b/docs/tutorials/snapshots.rst @@ -478,9 +478,13 @@ Rosbag Configuration Options - In-memory ring-buffer cap; oldest buffered messages drop once exceeded, so a broad subscribe set cannot grow memory without bound. * - ``snapshots.rosbag.format`` - - ``"sqlite3"`` - - Bag storage format: ``"sqlite3"`` (default, widely compatible) or - ``"mcap"`` (more efficient compression, requires plugin). + - ``"mcap"`` + - Bag storage format: ``"mcap"`` (default; opens directly in Foxglove and + Lichtblick) or ``"sqlite3"`` (also what an unknown format string lands + on). Neither is privileged: whichever is configured, an unavailable + plugin falls back automatically to the other one, and capture disables + itself only if neither loads. Both plugins are runtime dependencies of + the package. * - ``snapshots.rosbag.storage_path`` - ``""`` - Directory for bag files. Empty string uses system temp directory @@ -671,22 +675,19 @@ Saves resources but may miss context if fault confirms before buffer fills. .. note:: - The ``"mcap"`` format requires ``rosbag2_storage_mcap`` to be installed. - ``"sqlite3"`` (the default) is always shipped with rosbag2 and needs no extra - package. + ``"mcap"`` (the default) opens directly in Foxglove and Lichtblick with no + conversion step. ``"sqlite3"`` is also what an unknown format string lands + on. Both ``rosbag2_storage_mcap`` and ``rosbag2_storage_default_plugins`` + (which carries the sqlite3 plugin) are runtime dependencies of the package, + so both are installed alongside it, and neither is privileged over the + other. You do not have to switch formats manually: if a configured backend's plugin - is unavailable at startup, the FaultManager logs a warning naming the missing - package and automatically falls back to ``"sqlite3"`` for black-box capture. - If no storage backend is usable at all, rosbag capture self-disables (the node - keeps running and freeze-frame snapshots are unaffected) instead of crashing. - - To use mcap (e.g. for Foxglove), install the plugin: - - .. code-block:: bash - - # Install MCAP support (optional) - sudo apt install ros-${ROS_DISTRO}-rosbag2-storage-mcap + is somehow still unavailable at startup, the FaultManager logs a warning + naming the missing package and automatically falls back to the other + backend for black-box capture. If neither storage backend is usable at all, + rosbag capture self-disables (the node keeps running and freeze-frame + snapshots are unaffected) instead of crashing. Downloading Rosbag Files ^^^^^^^^^^^^^^^^^^^^^^^^ @@ -758,7 +759,6 @@ For production use with conservative resource usage: duration_after_sec: 0.5 topics: "config" # Use same topics as JSON snapshots lazy_start: true # Save resources until fault detected - format: "sqlite3" max_bag_size_mb: 25 max_total_storage_mb: 200 auto_cleanup: true @@ -781,7 +781,6 @@ For development with maximum context: duration_after_sec: 2.0 # 2 seconds after topics: "config" lazy_start: false # Always recording - format: "sqlite3" storage_path: "/var/log/ros2_medkit/rosbags" max_bag_size_mb: 100 max_total_storage_mb: 1000 diff --git a/src/ros2_medkit_fault_manager/CMakeLists.txt b/src/ros2_medkit_fault_manager/CMakeLists.txt index 836fa1484..58eb3cc63 100644 --- a/src/ros2_medkit_fault_manager/CMakeLists.txt +++ b/src/ros2_medkit_fault_manager/CMakeLists.txt @@ -114,6 +114,14 @@ if(BUILD_TESTING) include(ROS2MedkitTestDomain) + # Manifest test: the storage plugins declared as exec/test dependencies. + # Creates no ROS node, so it needs no domain. + find_package(ament_cmake_pytest REQUIRED) + ament_add_pytest_test(test_rosbag_storage_dependency test/test_rosbag_storage_dependency.py + TIMEOUT 60) + set_tests_properties(test_rosbag_storage_dependency PROPERTIES LABELS "unit") + medkit_test_needs_no_domain(test_rosbag_storage_dependency) + # Unit tests # Each GTest target that instantiates ROS 2 nodes takes a ROS_DOMAIN_ID from this # package's pool so it cannot interfere with launch_testing integration tests that run @@ -171,6 +179,11 @@ if(BUILD_TESTING) medkit_add_launch_test(test_rosbag_integration test/test_rosbag_integration.test.py TIMEOUT 90 LABELS "integration") + # Launches without snapshots.rosbag.format, so the parameter default (mcap) + # is what is under test rather than an operator's explicit choice. + medkit_add_launch_test(test_rosbag_default_format test/test_rosbag_default_format.test.py TIMEOUT 90 + LABELS "integration") + medkit_add_launch_test(test_entity_thresholds_integration test/test_entity_thresholds_integration.test.py TIMEOUT 60 LABELS "integration") diff --git a/src/ros2_medkit_fault_manager/README.md b/src/ros2_medkit_fault_manager/README.md index 5ada5ebb2..b325273fc 100644 --- a/src/ros2_medkit_fault_manager/README.md +++ b/src/ros2_medkit_fault_manager/README.md @@ -109,8 +109,25 @@ patterns: - /cmd_vel ``` +### Rosbag Capture + +Black-box rosbag capture (`snapshots.rosbag.*`) writes bags in `mcap` format by +default, which Foxglove and Lichtblick open directly with no conversion step. +`sqlite3` remains available through `snapshots.rosbag.format` and is also what an +unknown format string lands on. Neither backend is privileged over the other: +whichever is configured, if its plugin is unavailable at startup the FaultManager +logs a warning naming the missing package and falls back to the other one +automatically, disabling capture only if neither loads. Both `rosbag2_storage_mcap` +and `rosbag2_storage_default_plugins` (the sqlite3 plugin) are runtime dependencies +of this package, so a normal install pulls in both backends. See +`docs/config/fault-manager.rst` and `docs/tutorials/snapshots.rst` for the full +parameter list and recording lifecycle. + ### Storage Backends +This is the persisted fault store (`storage_type`), separate from the rosbag storage +format used by black-box capture (`snapshots.rosbag.format`, see Rosbag Capture above). + **SQLite (default)**: Faults are persisted to disk and survive node restarts. Uses WAL mode for optimal performance. **Memory**: Faults are stored in memory only. Useful for testing or when persistence is not required. diff --git a/src/ros2_medkit_fault_manager/config/snapshots.yaml b/src/ros2_medkit_fault_manager/config/snapshots.yaml index bfdb5fcbd..fa89b2f15 100644 --- a/src/ros2_medkit_fault_manager/config/snapshots.yaml +++ b/src/ros2_medkit_fault_manager/config/snapshots.yaml @@ -130,13 +130,18 @@ rosbag: # When false, ring buffer runs continuously from node startup lazy_start: false - # Bag file format (default: "sqlite3") + # Bag file format (default: "mcap") # Options: - # "sqlite3" - Default; always shipped with rosbag2, zero extra setup - # "mcap" - Opt-in; efficient and Foxglove-readable, needs rosbag2_storage_mcap - # If the selected plugin is not installed, capture automatically falls back to - # sqlite3, and disables itself only if no backend is usable (never crashes). - format: "sqlite3" + # "mcap" - Default; opens directly in Foxglove and Lichtblick + # "sqlite3" - Also what an unknown format string lands on + # Neither backend is privileged over the other: whichever is configured, if + # its plugin is unavailable at startup the FaultManager logs a warning + # naming the missing package and falls back to the other one automatically, + # disabling capture only if neither loads (never crashes). Both + # rosbag2_storage_mcap and rosbag2_storage_default_plugins (the sqlite3 + # plugin) are runtime dependencies of this package, so a normal install + # pulls in both backends. + format: "mcap" # Storage path for bag files (default: "" = system temp directory) # Empty string uses /tmp/rosbag_snapshots/ diff --git a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/snapshot_capture.hpp b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/snapshot_capture.hpp index 829a4a4d5..b14152036 100644 --- a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/snapshot_capture.hpp +++ b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/snapshot_capture.hpp @@ -64,9 +64,13 @@ struct RosbagConfig { /// If false (default), ring buffer runs continuously from startup bool lazy_start{false}; - /// Storage format: "sqlite3" (default, always shipped with rosbag2) or "mcap" - /// (opt-in; needs rosbag2_storage_mcap, falls back to sqlite3 if unavailable). - std::string format{"sqlite3"}; + /// Storage format: "mcap" (default; what Foxglove and Lichtblick open directly) + /// or "sqlite3" (also what an unknown format string lands on). Neither is + /// privileged over the other: whichever is configured, an unavailable plugin + /// falls back automatically to the other one, and capture disables itself + /// only if neither loads. Both plugins are runtime dependencies of this + /// package. + std::string format{"mcap"}; /// Path to store bag files (empty = system temp directory) std::string storage_path; diff --git a/src/ros2_medkit_fault_manager/package.xml b/src/ros2_medkit_fault_manager/package.xml index 723c7fed6..7c217f04f 100644 --- a/src/ros2_medkit_fault_manager/package.xml +++ b/src/ros2_medkit_fault_manager/package.xml @@ -25,18 +25,29 @@ launch launch_ros + + rosbag2_storage_mcap + rosbag2_storage_default_plugins + ament_lint_auto ament_lint_common - - rosbag2_storage_mcap ament_cmake_clang_format ament_cmake_clang_tidy ament_cmake_gtest + ament_cmake_pytest launch_ros launch_testing_ament_cmake launch_testing_ros + python3-pytest rosbag2_py + rosbag2_storage_mcap + rosbag2_storage_default_plugins sensor_msgs std_msgs diff --git a/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp b/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp index 72841cd6e..ab9628b30 100644 --- a/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp +++ b/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp @@ -1114,7 +1114,7 @@ SnapshotConfig FaultManagerNode::create_snapshot_config() { config.rosbag.qos_match = declare_parameter("snapshots.rosbag.qos_match", true); config.rosbag.lazy_start = declare_parameter("snapshots.rosbag.lazy_start", false); - config.rosbag.format = declare_parameter("snapshots.rosbag.format", "sqlite3"); + config.rosbag.format = declare_parameter("snapshots.rosbag.format", "mcap"); config.rosbag.storage_path = declare_parameter("snapshots.rosbag.storage_path", ""); int64_t max_bag_size = declare_parameter("snapshots.rosbag.max_bag_size_mb", 50); diff --git a/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp b/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp index 8bf22853b..6cd65ee28 100644 --- a/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp @@ -39,11 +39,14 @@ namespace { /// Actionable install hint for an unavailable storage backend. std::string storage_plugin_hint(const std::string & format) { + const char * distro = std::getenv("ROS_DISTRO"); + const std::string d = (distro && *distro) ? distro : "$ROS_DISTRO"; if (format == "mcap") { - const char * distro = std::getenv("ROS_DISTRO"); - const std::string d = (distro && *distro) ? distro : "$ROS_DISTRO"; return "install ros-" + d + "-rosbag2-storage-mcap"; } + if (format == "sqlite3") { + return "install ros-" + d + "-rosbag2-storage-default-plugins"; + } return "check the rosbag2 storage plugin installation"; } @@ -163,9 +166,12 @@ RosbagCapture::RosbagCapture(rclcpp::Node * node, FaultStorage * storage, const // Resolve a usable storage backend without ever terminating the FaultManager: // an unavailable plugin (e.g. rosbag2_storage_mcap not installed) must degrade, - // not crash. Unknown formats and a missing plugin fall back to sqlite3 (always - // shipped with rosbag2); if no backend works, disable capture and keep running - // (freeze-frame snapshots are independent of rosbag). + // not crash. An unknown format string falls back to sqlite3. Neither backend is + // privileged over the other: both mcap and sqlite3 are declared runtime + // dependencies of this package, so whichever format is CONFIGURED, if its + // plugin fails to load, resolution falls back to the OTHER one; if neither + // backend works, disable capture and keep running (freeze-frame snapshots are + // independent of rosbag). if (config_.format != "sqlite3" && config_.format != "mcap") { RCLCPP_WARN(node_->get_logger(), "Unknown rosbag storage format '%s'; using 'sqlite3'", config_.format.c_str()); config_.format = "sqlite3"; @@ -173,31 +179,25 @@ RosbagCapture::RosbagCapture(rclcpp::Node * node, FaultStorage * storage, const if (auto probe_err = storage_probe_(config_.format)) { const std::string reason = truncate_reason(*probe_err); - if (config_.format == "sqlite3") { - // The always-shipped baseline failed: an environment-level problem (broken - // rosbag2 base install / disk / permissions), not a missing optional plugin. - RCLCPP_WARN(node_->get_logger(), - "sqlite3 rosbag storage is unavailable (%s); the rosbag2 base install may be broken. " - "Black-box rosbag capture disabled (freeze-frame snapshots still work)", - reason.c_str()); - config_.enabled = false; - return; - } - if (auto sqlite_err = storage_probe_("sqlite3")) { + const std::string other_format = (config_.format == "sqlite3") ? "mcap" : "sqlite3"; + if (auto other_err = storage_probe_(other_format)) { RCLCPP_WARN(node_->get_logger(), - "No usable rosbag storage backend: '%s' (%s) and sqlite3 (%s) both failed to load; black-box " - "rosbag capture disabled (freeze-frame snapshots still work)", - config_.format.c_str(), reason.c_str(), truncate_reason(*sqlite_err).c_str()); + "No usable rosbag storage backend: '%s' (%s) and '%s' (%s) both failed to load; install " + "rosbag2_storage_mcap and rosbag2_storage_default_plugins. Black-box rosbag capture disabled " + "(freeze-frame snapshots still work)", + config_.format.c_str(), reason.c_str(), other_format.c_str(), truncate_reason(*other_err).c_str()); config_.enabled = false; return; } - // Explicitly-configured (non-default) format unavailable: name the fix so an - // operator who chose e.g. mcap for Foxglove is not silently downgraded. + // The configured format is unavailable, whether it is the default (mcap) or + // an explicit choice (including sqlite3): name the fix so an operator + // relying on it - Foxglove/Lichtblick via mcap, or legacy tooling via + // sqlite3 - is not silently downgraded without knowing what to install. RCLCPP_WARN(node_->get_logger(), - "Rosbag storage format '%s' is unavailable (%s); %s. Falling back to 'sqlite3' for black-box " - "capture (sqlite3 bags are not directly Foxglove-readable)", - config_.format.c_str(), reason.c_str(), storage_plugin_hint(config_.format).c_str()); - config_.format = "sqlite3"; + "Rosbag storage format '%s' is unavailable (%s); %s. Falling back to '%s' for black-box capture", + config_.format.c_str(), reason.c_str(), storage_plugin_hint(config_.format).c_str(), + other_format.c_str()); + config_.format = other_format; } RCLCPP_INFO(node_->get_logger(), "RosbagCapture initialized (duration=%.1fs, after=%.1fs, lazy_start=%s, format=%s)", diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp index c88b67f8c..d7b9d3f66 100644 --- a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp @@ -118,10 +118,28 @@ TEST_F(RosbagCaptureTest, ConstructorWithDisabledRosbag) { EXPECT_NO_THROW(RosbagCapture(node_.get(), storage_.get(), rosbag_config, snapshot_config)); } +// @verifies REQ_INTEROP_088 +TEST_F(RosbagCaptureTest, DefaultFormatIsMcap) { + // mcap is what Foxglove and Lichtblick open without conversion, and what + // rosbag2 itself defaults to since Iron. A silent drift back to sqlite3 would + // hand operators a black box their viewer cannot read. + RosbagConfig defaults; + EXPECT_EQ(defaults.format, "mcap"); +} + // @verifies REQ_INTEROP_088 TEST_F(RosbagCaptureTest, ConstructorFallsBackOnUnknownFormat) { - // An unknown/unavailable format must NOT terminate the node; it degrades to - // sqlite3 (always shipped with rosbag2) and capture stays enabled. + // An unknown format string must NOT terminate the node. It normalises to + // "sqlite3" first, then goes through the same real probe as an explicit + // sqlite3 configuration would, so this test uses the real probe on + // purpose - unlike the crash-safety tests below, which inject a double. + // That means the resolved format is not fixed: on a host with only mcap + // installed and not sqlite3 (the buildfarm chroot before + // rosbag2_storage_default_plugins lands as a real dependency), resolution + // falls back to mcap instead, which is the fallback being symmetric doing + // exactly its job, not a failure. The contract this test is actually about + // is that capture stays enabled and resolves to one of the two real + // backends, not "invalid_format" - not which one that happens to be here. auto rosbag_config = create_rosbag_config(); rosbag_config.format = "invalid_format"; auto snapshot_config = create_snapshot_config(); @@ -129,7 +147,8 @@ TEST_F(RosbagCaptureTest, ConstructorFallsBackOnUnknownFormat) { EXPECT_NO_THROW(rb = std::make_shared(node_.get(), storage_.get(), rosbag_config, snapshot_config)); ASSERT_NE(rb, nullptr); EXPECT_TRUE(rb->is_enabled()); - EXPECT_EQ(rb->config().format, "sqlite3"); + EXPECT_TRUE(rb->config().format == "sqlite3" || rb->config().format == "mcap") + << "resolved format was '" << rb->config().format << "', neither a real backend"; } // The crash-safety branches below force the storage probe via an injected double, @@ -156,6 +175,29 @@ TEST_F(RosbagCaptureTest, ConfiguredFormatUnavailableFallsBackToSqlite3) { EXPECT_EQ(rb->config().format, "sqlite3"); } +// @verifies REQ_INTEROP_088 +TEST_F(RosbagCaptureTest, ConfiguredSqlite3UnavailableFallsBackToMcap) { + // The symmetric case of the fallback above: a known format (sqlite3) whose + // plugin is unavailable degrades to mcap and capture stays enabled. Neither + // backend is privileged - whichever is configured falls back to the other, + // not always to sqlite3. + auto rosbag_config = create_rosbag_config(); + rosbag_config.format = "sqlite3"; + auto snapshot_config = create_snapshot_config(); + RosbagCapture::StorageProbeFn probe = [](const std::string & f) -> std::optional { + if (f == "sqlite3") { + return std::string("simulated: sqlite3 plugin not found"); + } + return std::nullopt; // mcap usable + }; + std::shared_ptr rb; + EXPECT_NO_THROW( + rb = std::make_shared(node_.get(), storage_.get(), rosbag_config, snapshot_config, probe)); + ASSERT_NE(rb, nullptr); + EXPECT_TRUE(rb->is_enabled()); + EXPECT_EQ(rb->config().format, "mcap"); +} + // @verifies REQ_INTEROP_088 TEST_F(RosbagCaptureTest, NoUsableBackendDisablesCaptureWithoutCrashing) { // When neither the configured format nor sqlite3 is usable, capture self-disables @@ -175,8 +217,10 @@ TEST_F(RosbagCaptureTest, NoUsableBackendDisablesCaptureWithoutCrashing) { // @verifies REQ_INTEROP_088 TEST_F(RosbagCaptureTest, Sqlite3BaselineUnavailableDisablesCapture) { - // When the always-shipped sqlite3 baseline itself fails to load, capture - // self-disables rather than crashing the node. + // When sqlite3 is configured and neither it nor its mcap fallback loads, + // capture self-disables rather than crashing the node - the symmetric case + // of NoUsableBackendDisablesCaptureWithoutCrashing above, starting from + // sqlite3 instead of mcap. auto rosbag_config = create_rosbag_config(); rosbag_config.format = "sqlite3"; auto snapshot_config = create_snapshot_config(); diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py b/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py new file mode 100644 index 000000000..4946eded8 --- /dev/null +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py @@ -0,0 +1,362 @@ +#!/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. + +""" +What an operator who configured nothing actually gets on disk. + +test_rosbag_integration.test.py sets `snapshots.rosbag.format` explicitly and +pins it. This launches the fault manager WITHOUT that parameter, so the +struct/parameter default is what is under test, not an operator's choice. +""" + +import os +import tempfile +import time +import unittest + +from launch import LaunchDescription +import launch.actions +import launch_ros.actions +import launch_testing.actions +import launch_testing.markers +import rclpy +from rclpy.node import Node +from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy +from ros2_medkit_msgs.msg import Fault +from ros2_medkit_msgs.srv import GetRosbag, ReportFault +from sensor_msgs.msg import Temperature + + +def get_coverage_env(): + """Get environment variables for gcov coverage data collection.""" + try: + from ament_index_python.packages import get_package_prefix + pkg_prefix = get_package_prefix('ros2_medkit_fault_manager') + workspace = os.path.dirname(os.path.dirname(pkg_prefix)) + build_dir = os.path.join(workspace, 'build', 'ros2_medkit_fault_manager') + + if os.path.exists(build_dir): + return { + 'GCOV_PREFIX': build_dir, + 'GCOV_PREFIX_STRIP': str(build_dir.count(os.sep)), + } + except Exception: + # Coverage environment is optional; on any error, fall back to no extra coverage config + pass + return {} + + +# Create a temp directory for rosbag storage that persists for test duration +ROSBAG_STORAGE_PATH = tempfile.mkdtemp(prefix='rosbag_default_format_test_') + +# Path to temp publisher script (set in generate_test_description, cleaned up in shutdown test) +PUBLISHER_SCRIPT_PATH = None + + +def generate_test_description(): + """Generate launch description with fault_manager node with rosbag enabled.""" + # Use Python script for publishers to ensure proper topic type registration + # ros2 topic pub has issues with topic type discovery in some configurations + publisher_script = """ +import rclpy +from rclpy.node import Node +from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy +from sensor_msgs.msg import Temperature +from std_msgs.msg import String +import time + +class TestPublisher(Node): + def __init__(self): + super().__init__('test_publisher') + import os + domain_id = os.environ.get('ROS_DOMAIN_ID', 'not set') + self.get_logger().info(f'Using ROS_DOMAIN_ID: {domain_id}') + + # Use sensor data QoS to match rosbag capture subscriptions + qos = QoSProfile( + reliability=ReliabilityPolicy.BEST_EFFORT, + history=HistoryPolicy.KEEP_LAST, + depth=10 + ) + self.temp_pub = self.create_publisher(Temperature, '/test/temperature', qos) + self.string_pub = self.create_publisher(String, '/test/status', qos) + self.timer = self.create_timer(0.1, self.publish) + self.counter = 0 + self.get_logger().info('TestPublisher started') + + def publish(self): + temp_msg = Temperature() + temp_msg.temperature = 25.0 + self.counter * 0.1 + temp_msg.variance = 0.1 + self.temp_pub.publish(temp_msg) + + string_msg = String() + string_msg.data = f'status_{self.counter}' + self.string_pub.publish(string_msg) + self.counter += 1 + if self.counter % 50 == 0: + self.get_logger().info(f'Published {self.counter} messages') + # Log available topics and types for debugging + topics = self.get_topic_names_and_types() + topic_info = [(t, types) for t, types in topics if 'test' in t] + if topic_info: + self.get_logger().info(f'Test topics visible: {topic_info}') + +def main(): + rclpy.init() + node = TestPublisher() + try: + rclpy.spin(node) + except KeyboardInterrupt: + pass + finally: + node.destroy_node() + rclpy.shutdown() + +if __name__ == '__main__': + main() +""" + + # Write publisher script to temp file and run it + # Store path in global for cleanup in post-shutdown test + global PUBLISHER_SCRIPT_PATH + script_file = tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) + script_file.write(publisher_script) + script_file.close() + PUBLISHER_SCRIPT_PATH = script_file.name + + # Inherit the ROS_DOMAIN_ID the domain wrapper allocated for this test, so + # it is isolated from everything else running in parallel. + # ROS_LOCALHOST_ONLY keeps discovery on loopback. + env = os.environ.copy() + env['ROS_LOCALHOST_ONLY'] = '1' + + test_publisher = launch.actions.ExecuteProcess( + cmd=['python3', script_file.name], + output='screen', + name='test_publisher', + env=env, + ) + + fault_manager_env = get_coverage_env() + fault_manager_env['ROS_LOCALHOST_ONLY'] = '1' + + fault_manager_node = launch_ros.actions.Node( + package='ros2_medkit_fault_manager', + executable='fault_manager_node', + name='fault_manager', + output='screen', + additional_env=fault_manager_env, + parameters=[{ + 'storage_type': 'memory', + 'confirmation_threshold': -1, # Single report confirms immediately + # Rosbag configuration. Deliberately no 'snapshots.rosbag.format' here: + # the default is what this test verifies. + 'snapshots.rosbag.enabled': True, + 'snapshots.rosbag.duration_sec': 2.0, # 2 second buffer + 'snapshots.rosbag.duration_after_sec': 0.5, # 0.5 second after confirm + # Use explicit topics - faster than discovery, more reliable for tests + 'snapshots.rosbag.topics': '/test/temperature,/test/status', + 'snapshots.rosbag.storage_path': ROSBAG_STORAGE_PATH, + 'snapshots.rosbag.max_bag_size_mb': 10, + 'snapshots.rosbag.max_total_storage_mb': 50, + 'snapshots.rosbag.auto_cleanup': True, + # lazy_start=false: Start recording immediately + 'snapshots.rosbag.lazy_start': False, + }], + # Give the node room to flush coverage data at shutdown before SIGKILL. + sigterm_timeout='30', + sigkill_timeout='15', + ) + + # Delay fault_manager start so test_publisher has time to register topics + # DDS discovery can take several seconds to propagate topic types + delayed_fault_manager = launch.actions.TimerAction( + period=8.0, + actions=[fault_manager_node], + ) + + return ( + LaunchDescription([ + # Start publisher node first + test_publisher, + # Start fault_manager after delay + delayed_fault_manager, + launch_testing.actions.ReadyToTest(), + ]), + { + 'fault_manager_node': fault_manager_node, + 'test_publisher': test_publisher, + }, + ) + + +class TestRosbagDefaultFormat(unittest.TestCase): + """A fault manager launched with no format parameter (@verifies REQ_INTEROP_088).""" + + @classmethod + def setUpClass(cls): + """Initialize ROS 2 context and create service clients.""" + # Match the launch processes: inherit the CMake-injected + # ROS_DOMAIN_ID and keep discovery on loopback. + os.environ['ROS_LOCALHOST_ONLY'] = '1' + rclpy.init() + cls.node = Node('test_rosbag_default_format_client') + + # Create service clients + cls.report_fault_client = cls.node.create_client( + ReportFault, '/fault_manager/report_fault' + ) + cls.get_rosbag_client = cls.node.create_client( + GetRosbag, '/fault_manager/get_rosbag' + ) + + # Wait for services to be available + assert cls.report_fault_client.wait_for_service(timeout_sec=15.0), \ + 'report_fault service not available' + assert cls.get_rosbag_client.wait_for_service(timeout_sec=15.0), \ + 'get_rosbag service not available' + + # Wait for the background publisher to come up before letting the rosbag + # ring buffer fill (duration_sec=2.0 configured), so a slow publisher + # start does not leave the buffer empty. + deadline = time.time() + 15.0 + while (not cls.node.get_publishers_info_by_topic('/test/temperature') + and time.time() < deadline): + time.sleep(0.2) + time.sleep(3.0) + + @classmethod + def tearDownClass(cls): + """Shutdown ROS 2 context.""" + cls.node.destroy_node() + rclpy.shutdown() + + def _call_service(self, client, request, timeout_sec=10.0): + """Call a service and wait for response.""" + future = client.call_async(request) + rclpy.spin_until_future_complete(self.node, future, timeout_sec=timeout_sec) + self.assertIsNotNone(future.result(), 'Service call timed out') + return future.result() + + def _report_fault(self, fault_code, description='Test fault'): + """Report a fault and return the response.""" + request = ReportFault.Request() + request.fault_code = fault_code + request.event_type = ReportFault.Request.EVENT_FAILED + request.severity = Fault.SEVERITY_ERROR + request.description = description + request.source_id = '/test_node' + return self._call_service(self.report_fault_client, request) + + def _wait_for_rosbag(self, fault_code, timeout=12.0): + """ + Poll GetRosbag until the async post-fault recording is written. + + The recording plus its post-fault flush (duration_after_sec) can take + longer than a fixed sleep under sanitizer/coverage load, so a one-shot + call flakes. Mirror the polling loop in test_rosbag_entity_scope. + """ + request = GetRosbag.Request() + request.fault_code = fault_code + deadline = time.time() + timeout + response = self._call_service(self.get_rosbag_client, request) + while time.time() < deadline: + if response is not None and response.success: + return response + time.sleep(0.5) + response = self._call_service(self.get_rosbag_client, request) + return response + + def _wait_for_buffered_data(self, count=5, timeout=8.0): + """ + Wait until the rosbag ring buffer holds fresh data before a confirmation. + + Mirror of the same helper in test_rosbag_integration.test.py: subscribe to + the same source the fault manager records and wait for messages to arrive, + rather than sleeping a fixed amount. + """ + received = 0 + + def _cb(_msg): + nonlocal received + received += 1 + + # Match the publisher's BEST_EFFORT sensor QoS, else no messages arrive. + qos = QoSProfile( + reliability=ReliabilityPolicy.BEST_EFFORT, + history=HistoryPolicy.KEEP_LAST, + depth=10, + ) + sub = self.node.create_subscription(Temperature, '/test/temperature', _cb, qos) + try: + deadline = time.time() + timeout + while received < count and time.time() < deadline: + rclpy.spin_once(self.node, timeout_sec=0.1) + finally: + self.node.destroy_subscription(sub) + return received >= count + + def test_01_default_format_is_mcap(self): + """ + A fault manager launched with no format parameter writes an mcap bag. + + The unit case pins the struct default; this pins what an operator who + configured nothing actually gets on disk, which is the thing the docs + and the README promise. + """ + fault_code = 'DEFAULT_FORMAT_TEST' + + # Buffer should already have messages from background publishers. + self.assertTrue(self._wait_for_buffered_data(), + 'ring buffer never refilled after startup') + response = self._report_fault(fault_code, 'Default rosbag format test fault') + self.assertTrue(response.accepted) + + # Poll until the async post-fault recording is written. + rosbag_response = self._wait_for_rosbag(fault_code) + self.assertTrue(rosbag_response.success, + f'GetRosbag failed: {rosbag_response.error_message}') + self.assertEqual(rosbag_response.format, 'mcap') + + contents = os.listdir(rosbag_response.file_path) + self.assertTrue( + any(name.endswith('.mcap') for name in contents), + f'no .mcap file in the bag written with the default format: {contents}', + ) + + +@launch_testing.post_shutdown_test() +class TestRosbagDefaultFormatShutdown(unittest.TestCase): + """Post-shutdown tests.""" + + def test_exit_code(self, proc_info): + """Verify fault_manager exits cleanly.""" + launch_testing.asserts.assertExitCodes( + proc_info, + process='fault_manager_node' + ) + + def test_cleanup_temp_directory(self): + """Clean up temporary rosbag storage directory and publisher script.""" + import shutil + if os.path.exists(ROSBAG_STORAGE_PATH): + shutil.rmtree(ROSBAG_STORAGE_PATH, ignore_errors=True) + print(f'Cleaned up temp directory: {ROSBAG_STORAGE_PATH}') + + # Clean up the temporary publisher script + if PUBLISHER_SCRIPT_PATH and os.path.exists(PUBLISHER_SCRIPT_PATH): + os.unlink(PUBLISHER_SCRIPT_PATH) + print(f'Cleaned up temp script: {PUBLISHER_SCRIPT_PATH}') diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_storage_dependency.py b/src/ros2_medkit_fault_manager/test/test_rosbag_storage_dependency.py new file mode 100644 index 000000000..70cbea674 --- /dev/null +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_storage_dependency.py @@ -0,0 +1,123 @@ +# Copyright 2026 bburda +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +The manifest claim that black-box capture works on a stock install. + +The plugin that provides a rosbag storage format lives in its own package. +`rosbag2_storage` is only the plugin interface and `libsqlite3-dev` is the C +library the fault database uses, so a manifest naming those two and stopping +leaves every storage plugin out of the install closure. What the user gets then +is not an error: RosbagCapture logs a warning and disables itself, so the feature +the README leads with is silently off. + +Observed on the ROS build farm, whose chroot carries exactly `rosbag2_storage` +and `rosbag2_storage_mcap`, where `test_rosbag_capture` fails on +`Could not load/open plugin with storage id 'sqlite3'`. + +Both plugins are load-bearing. mcap is the default the bags are written in. +An unknown format string still normalizes to sqlite3, but an unavailable +plugin falls back to whichever of the two is *not* configured - neither +backend is privileged over the other - and capture disables itself only +when neither one loads. +""" + +import pathlib +import sys +import xml.etree.ElementTree as ElementTree + +import pytest + +MANIFEST = pathlib.Path(__file__).resolve().parent.parent / 'package.xml' + +# The default backend. One name on every distro we release for. +MCAP_PLUGIN = 'rosbag2_storage_mcap' +# The sqlite3 backend, and what an unknown format string normalizes to. An +# *unavailable* configured format falls back to whichever plugin is not +# configured, not to this one specifically. On humble this name is the +# sqlite3 plugin itself; on jazzy and lyrical it is a metapackage that pulls +# sqlite3 together with mcap. +SQLITE3_PLUGIN = 'rosbag2_storage_default_plugins' + + +def declared(*tags): + """Return the dependency names the manifest declares under the given tags.""" + root = ElementTree.parse(MANIFEST).getroot() + names = set() + for tag in tags: + names.update(element.text.strip() for element in root.findall(tag)) + return names + + +def declared_elements(*tags): + """Return the manifest elements under the given tags, not just their text.""" + root = ElementTree.parse(MANIFEST).getroot() + elements = [] + for tag in tags: + elements.extend(root.findall(tag)) + return elements + + +@pytest.mark.parametrize('plugin', [MCAP_PLUGIN, SQLITE3_PLUGIN]) +def test_storage_plugin_is_an_exec_dependency(plugin): + """An installed fault manager can open the formats it writes and falls back to.""" + names = declared('exec_depend', 'depend') + assert plugin in names, ( + f'{plugin} is not an exec dependency, so a package install pulls in no such ' + f'rosbag storage plugin and black-box capture degrades or disables itself. ' + f'Declared: {sorted(names)}' + ) + + +@pytest.mark.parametrize('plugin', [MCAP_PLUGIN, SQLITE3_PLUGIN]) +def test_storage_plugin_is_a_test_dependency(plugin): + """ + The test chroot gets the plugins too, so the rosbag tests mean something. + + An exec dependency alone is not installed at build time, and build time is + when the ROS build farm runs the tests. + """ + names = declared('test_depend', 'depend') + assert plugin in names, ( + f'{plugin} is not a test dependency, so the tests that assert rosbag capture ' + f'is enabled run without the plugin they need. Declared: {sorted(names)}' + ) + + +@pytest.mark.parametrize('plugin', [MCAP_PLUGIN, SQLITE3_PLUGIN]) +def test_storage_plugin_dependency_has_no_condition(plugin): + """ + A conditional dependency can pass every assertion above and still ship broken. + + ``rosbag2_storage_mcap`` + still reads back as declared by ``declared()`` above - the tag and its text + are untouched - but rosdep resolves nothing for it on the excluded distro, + so the plugin is silently absent there exactly the way the missing + declaration was before this task, just scoped to one distro instead of all + of them. Covers both the ``exec_depend`` and ``test_depend`` declaration for + each plugin, four in total. + """ + offenders = [ + ElementTree.tostring(element, encoding='unicode').strip() + for element in declared_elements('exec_depend', 'test_depend', 'depend') + if element.text and element.text.strip() == plugin and element.get('condition') is not None + ] + assert not offenders, ( + f'{plugin} is declared with a condition attribute, which can silently drop it ' + f'from the install closure on the distro(s) the condition excludes: {offenders}' + ) + + +if __name__ == '__main__': + sys.exit(pytest.main([__file__, '-v'])) diff --git a/src/ros2_medkit_gateway/README.md b/src/ros2_medkit_gateway/README.md index 39933e1ac..520d85256 100644 --- a/src/ros2_medkit_gateway/README.md +++ b/src/ros2_medkit_gateway/README.md @@ -1044,7 +1044,7 @@ Rosbag capture is configured via FaultManager parameters. See `config/snapshots. | `snapshots.rosbag.topics` | string | `"entity"` | Topic selection: `"entity"` (default; faulting node's topics + `/tf`), `"config"`, `"all"`, or `"explicit"` | | `snapshots.rosbag.exclude_sensor_topics` | bool | `true` | Auto-exclude image/points/depth/compressed in broad modes (`include_topics` re-adds) | | `snapshots.rosbag.qos_match` | bool | `true` | Match each topic's publisher QoS for faithful capture | -| `snapshots.rosbag.format` | string | `"sqlite3"` | Bag format: `"sqlite3"` or `"mcap"` | +| `snapshots.rosbag.format` | string | `"mcap"` | Bag format: `"mcap"` (default; opens directly in Foxglove and Lichtblick) or `"sqlite3"`. Neither is privileged - an unavailable plugin falls back automatically to the other one | | `snapshots.rosbag.auto_cleanup` | bool | `true` | Delete bag when fault is cleared | | `snapshots.rosbag.max_buffer_mb` | int | `256` | Ring-buffer RAM cap (oldest messages drop past it) | | `snapshots.rosbag.max_bag_size_mb` | int | `50` | Max size per bag file | diff --git a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp index e89268ee6..f36f20a13 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp @@ -238,8 +238,8 @@ BulkDataHandlers::list_descriptors(const http::TypedRequest & req) { dto::Collection response; for (const auto & rosbag : all_rosbags) { std::string fault_code = rosbag.value("fault_code", ""); - // Default to sqlite3 (the FaultManager default) when a bag predates the - // persisted format field; the per-bag metadata normally carries the real one. + // Default to sqlite3 (the historical FaultManager default) when a bag predates + // the persisted format field; the per-bag metadata normally carries the real one. std::string format = rosbag.value("format", "sqlite3"); uint64_t size_bytes = rosbag.value("size_bytes", uint64_t{0}); double duration_sec = rosbag.value("duration_sec", 0.0); @@ -365,8 +365,8 @@ http::Result BulkDataHandlers::download(const http::TypedR } std::string file_path = rosbag_result.data["file_path"].get(); - // Default to sqlite3 (the FaultManager default) for bags predating the format - // field; metadata normally carries the real one persisted at capture time. + // Default to sqlite3 (the historical FaultManager default) for bags predating the + // format field; metadata normally carries the real one persisted at capture time. std::string format = rosbag_result.data.value("format", "sqlite3"); mimetype = get_rosbag_mimetype(format); filename = fault_code + "." + format; From 0897a9b87b38ea3481011ccb44a81ce3cd9b4c43 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Wed, 12 Aug 2026 17:20:05 +0200 Subject: [PATCH 2/4] fix(fault_manager): serialize the rosbag storage-plugin loader process-wide `rosbag2_cpp::Writer` loads its storage plugin inside `open()`, and `~Writer` can unload it. The racing state is class_loader's process-global library registry, so the failure is a double unload of one shared handle: the winner calls `dlclose` and zeroes the `rcutils_shared_library_t`, the loser's `dlclose` returns `shared object not open` and it then calls the now-null `allocator.deallocate`. The faulting frame is `rcutils_unload_shared_library`. Two threads doing nothing but open and close a writer on one storage id crashed in 20 of 20 runs, on mcap and on sqlite3 alike, so this is not specific to either backend. It reproduces on the merge base as well. A process-wide mutex now serializes construction, `open()` and destruction, including inside the storage probe, which took no lock at all. A control with `open()` outside that lock still leaked failures, which is why it is inside. `writer_mutex_` keeps guarding writer data access and is no longer held across an open or a close, so a close cannot stall the post-roll callbacks. Every destruction site moves the writer out under `writer_mutex_` and destroys it under the plugin mutex; the reverse order closes a deadlock cycle with `open_bag_writer`. The mutex is deliberately leaked, because one destroyed at static-destruction time while another thread is closing a writer reopens the window. Measured: 140 of 140 runs clean over 112000 operations per backend, against a baseline of 0 of 20. Close under the lock costs about 0.37 ms for an empty bag and about 1.1 ms for a 256 MB one, the largest a single flush can produce. The earlier reasoning that moved the close out of the lock was right about the I/O cost and wrong about where it was safe to pay it, so both halves are kept. --- .../design/index.rst | 64 +++++- .../rosbag_capture.hpp | 35 ++- .../src/rosbag_capture.cpp | 210 +++++++++++++++--- .../test/test_rosbag_capture.cpp | 154 +++++++++++++ .../test/test_rosbag_integration.test.py | 34 ++- 5 files changed, 452 insertions(+), 45 deletions(-) diff --git a/src/ros2_medkit_fault_manager/design/index.rst b/src/ros2_medkit_fault_manager/design/index.rst index 42641f8d5..bf6423ae6 100644 --- a/src/ros2_medkit_fault_manager/design/index.rst +++ b/src/ros2_medkit_fault_manager/design/index.rst @@ -336,13 +336,67 @@ critical section: the guard, the start time and the writer. Releasing the guard reaching for ``writer_mutex_`` afterwards leaves a gap in which the incoming confirmation installs its writer and the outgoing finalise then destroys it, after which the new recording writes through a null pointer - every message dropped - and still stores a row -for the empty bag it produced. The writer is only *closed* outside the locks, once it is -exclusively the finalise's own, because flushing a bag is real I/O. +for the empty bag it produced. The writer is *closed* only after all three of +``post_fault_timer_mutex_``, ``capture_topics_mutex_`` and ``writer_mutex_`` are released, +once it is exclusively the finalise's own, because flushing a bag is real I/O. It is not +closed unsynchronised, though. It is closed under a different lock, and that lock exists +for a reason that has nothing to do with this class's state. + +Closing a bag, and the lock that is not the data lock +""""""""""""""""""""""""""""""""""""""""""""""""""""" + +``~Writer`` can unload the storage plugin's shared library and ``open()`` loads it, and the +state both touch is **process-global**: ``class_loader`` keeps one registry of loaded +libraries keyed by library path, so every writer of a given format shares one +``rcutils_shared_library_t`` no matter which thread, which ``RosbagCapture`` or which +``FaultManagerNode`` created it. A close racing an open puts two threads into ``dlclose`` +on the same handle. The winner closes it and zeroes the struct; the loser's ``dlclose`` +reports "shared object not open" and then calls the now-null ``allocator.deallocate``, +and the process dies in ``rcutils_unload_shared_library`` with the instruction pointer at +zero. Reproduced outside this package with threads doing nothing but open/close loops on +one format: four threads x 200 iterations failed in 20 of 20 runs on ``mcap`` and in 20 of +20 on ``sqlite3``, so it belongs to neither backend and to no single instance. + +An instance member cannot serialise process-global state, so the lock is a file-scope +``plugin_mutex()`` in ``rosbag_capture.cpp``, held across a writer's construction, its +``open()`` and its destruction, on every path that has one: both finalise paths, +``discard_active_writer()``, the destructor, and the storage-backend probe that every +capture runs while it is being constructed. ``open()`` is inside the lock deliberately - +the constructor reaches no loader, the open is what loads the plugin, and a lock around +construction and destruction alone still let failures through. The mutex is leaked on +purpose (a ``new std::mutex`` that is never deleted): a static mutex destroyed during +static destruction, while another thread is closing a bag, reopens the very window it +exists to close. With it, the same loops ran 0 failures in 112000 operations on each +backend. + +It is deliberately **not** ``writer_mutex_``. That lock is taken by ``message_callback()`` +for every message of a post-roll and by the flush loop for every buffered message, so +charging a close to it would stall the capture's own write path for the length of a flush +plus a ``metadata.yaml`` write. Keeping the close off the data lock was the original +design's call, and the cost it avoided is real; what it left unpaid was safety. Measured on +one workstation, single-threaded: a close costs about 0.37 ms for a bag with no messages +(200 samples) and about 1.1 ms for one holding 256 MB, the ring buffer's default RAM cap, +split at the default 50 MB per file (12 samples). A recording opening at that moment waits +behind that. The figure is small because closing flushes to the page cache and writes +``metadata.yaml``; it does not ``fsync``, so slow or synchronous storage will cost more. The resulting order is ``node rosbag mutex -> post_fault_timer_mutex_ -> -{capture_topics_mutex_, writer_mutex_}``, with ``buffer_mutex_`` never held across another -lock. Paths that take the capture-topics or writer locks on their own release each before -taking the next, so no reverse edge exists and the order is acyclic. +{capture_topics_mutex_, writer_mutex_}`` plus ``plugin_mutex() -> writer_mutex_``, with +``buffer_mutex_`` never held across another lock. ``plugin_mutex()`` is taken in three +shapes: alone, to destroy a writer already moved out of ``active_writer_``; alone, across a +probe writer's construction, ``open()`` and destruction in ``default_storage_probe()``, +which never touches ``active_writer_`` at all; or as the outer of the pair in +``open_bag_writer()``. No path takes it while holding a lock of this class. That is what +fixes the shape every destruction site shares - move the writer out of ``active_writer_`` +under ``writer_mutex_``, release ``writer_mutex_``, then destroy under ``plugin_mutex()``. +Resetting in place under ``writer_mutex_`` would add the reverse edge and deadlock against +a concurrent ``open_bag_writer()``. That cycle is reachable precisely because the close sits +outside ``post_fault_timer_mutex_``: a confirmation running at the same time is not held at +the attach check, so it can be inside ``open_bag_writer()`` holding the plugin lock while +the finalise holds ``writer_mutex_`` and asks for it. Moving the close back inside +``post_fault_timer_mutex_`` would remove the cycle and reintroduce the stall the design +declines to pay for. Paths that take the capture-topics or writer locks on their own release +each before taking the next, so no other reverse edge exists and the order is acyclic. Honest durations """""""""""""""" diff --git a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/rosbag_capture.hpp b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/rosbag_capture.hpp index f859a9ac0..1e655418c 100644 --- a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/rosbag_capture.hpp +++ b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/rosbag_capture.hpp @@ -365,13 +365,18 @@ class RosbagCapture { /// node rosbag mutex -> post_fault_timer_mutex_ -> capture_topics_mutex_ /// node rosbag mutex -> post_fault_timer_mutex_ -> writer_mutex_ /// node rosbag mutex -> post_fault_timer_mutex_ -> node_ops_mutex_ + /// plugin_mutex() (file-scope, in rosbag_capture.cpp) -> writer_mutex_ /// buffer_mutex_ is never held across another lock. The paths that take /// capture_topics_mutex_ or writer_mutex_ on their own (the flush loop, the /// post-roll write path) release each before taking the next, so they add no - /// reverse edge. Everything that hands the RECORDING over - the guard, the - /// start time, the writer - must happen inside one post_fault_timer_mutex_ - /// critical section, or a confirmation racing a finalise ends up owning half of - /// the previous recording's state. + /// reverse edge. plugin_mutex() is taken either alone, to close a writer already + /// moved out of active_writer_, or as the OUTER of the pair in open_bag_writer(); + /// no path takes it while holding any lock of this class, and the finalise paths + /// close their writer only after post_fault_timer_mutex_, capture_topics_mutex_ + /// and writer_mutex_ are all released. Everything that hands the RECORDING over - + /// the guard, the start time, the writer - must happen inside one + /// post_fault_timer_mutex_ critical section, or a confirmation racing a finalise + /// ends up owning half of the previous recording's state. mutable std::mutex post_fault_timer_mutex_; rclcpp::TimerBase::SharedPtr post_fault_timer_; std::atomic recording_post_fault_{false}; @@ -399,6 +404,28 @@ class RosbagCapture { /// Active writer for current bag (kept open during post-fault recording) std::unique_ptr active_writer_; + /// Guards DATA ACCESS to active_writer_ and created_topics_, and nothing else. + /// Held only for a pointer handover or for one create_topic()/write() call, so + /// the two paths that take it in the hot path - message_callback() during a + /// post-roll and the flush loop - never wait behind bag I/O. + /// + /// It deliberately does NOT cover a Writer's construction, its open() or its + /// destruction. Those reach rosbag2's storage-plugin loader, whose racing state + /// is process-global rather than per capture, so an instance member cannot + /// exclude the thread that matters. They are serialised by plugin_mutex() in + /// rosbag_capture.cpp instead; see its definition for the crash, the reason it is + /// leaked, and the measurements. + /// + /// Order with respect to that lock: plugin_mutex() -> writer_mutex_. Only + /// open_bag_writer() holds both, and it takes them in that order. Every + /// destruction site for a writer that was active_writer_ therefore has one + /// shape - move the writer out of active_writer_ under writer_mutex_, RELEASE + /// writer_mutex_, then destroy it under plugin_mutex() - because closing in + /// place under writer_mutex_ would add the reverse edge and deadlock against a + /// concurrent open. default_storage_probe()'s writer is a separate case: it is + /// local, never becomes active_writer_, and so is constructed, opened and + /// destroyed under plugin_mutex() alone, without writer_mutex_ ever entering + /// the picture. std::mutex writer_mutex_; std::set created_topics_; diff --git a/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp b/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp index 6cd65ee28..b5d2f0bbd 100644 --- a/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp @@ -19,10 +19,14 @@ #include #include +#include #include +#include #include #include #include +#include +#include #include #include #include @@ -83,6 +87,68 @@ double span_sec_since(int64_t started_at_steady_ns) { return span > 0.0 ? span : 0.0; } +/// Serialises everything that reaches rosbag2's storage-plugin loader: a Writer's +/// construction, its open(), and its destruction. Nothing else belongs under it. +/// +/// The racing state is PROCESS-GLOBAL, not per capture, which is why this is not a member. +/// class_loader keeps ONE registry of loaded libraries keyed by library path +/// (class_loader::impl::getLoadedLibraryVector()), so every Writer of a given format shares +/// one rcutils_shared_library_t handle regardless of which object, thread or RosbagCapture +/// instance opened it. Let one Writer's destructor unload that library while another loads +/// it and both run dlclose on the same handle: the winner closes it and zeroes the struct, +/// the loser's dlclose fails with "shared object not open" and then calls the now-null +/// allocator.deallocate. The fault lands in rcutils_unload_shared_library with RIP=0. +/// +/// Measured standalone, outside this package, with threads doing nothing but open()/close() +/// on one format: 4 threads x 200 iterations crashed in 20 of 20 runs on mcap and failed in +/// 20 of 20 on sqlite3 - 15 of those ended in SIGSEGV and 2 more in an uncaught +/// class_loader::LibraryUnloadException reaching std::terminate (SIGABRT, not a segfault); +/// the remaining 3 finished all 200 iterations without a fatal end, but still logged a caught +/// plugin-load exception along the way, as did 13 of the 17 that crashed - 16 of the 20 +/// sqlite3 runs logged at least one such exception in total. Neither backend's bug. The same +/// loops with this lock held over construction, open() and destruction ran 0 failures in +/// 112000 operations per backend. +/// +/// open() has to be INSIDE the lock, not just the constructor and the destructor: the +/// constructor touches no loader, open() is what loads the plugin. Measured the same way, a +/// lock around construction and destruction only still leaked 2 failures in 32000 +/// operations. +/// +/// What it costs is what a concurrent open_bag_writer() waits behind, measured on one +/// workstation single-threaded: a close is ~0.37 ms for a bag with no messages (200 +/// samples) and ~1.1 ms for a 256 MB one split at the default 50 MB per file (12 samples), +/// on top of a ~1.0-1.4 ms open. Closing flushes to the page cache and writes +/// metadata.yaml without fsync, which is why the size barely moves the figure; slow or +/// synchronous storage will cost more. +/// +/// Deliberately never destroyed. A function-local static std::mutex would be destroyed +/// during static destruction, and a thread closing a bag at exit would then lock a destroyed +/// mutex, which reopens the exact window this exists to close. One leaked mutex for the life +/// of the process is the price. +/// +/// Process-wide as far as this translation unit reaches: rosbag_capture.cpp is compiled once +/// into fault_manager_lib, so every RosbagCapture in the process shares this mutex. It +/// cannot cover a Writer opened by code outside this package. +std::mutex & plugin_mutex() { + static std::mutex * m = new std::mutex; + return *m; +} + +/// Destroy a writer already taken out of active_writer_, under the plugin lock. +/// +/// Every destruction site in this file goes through here and all of them have the same +/// shape: move the writer out of the member under writer_mutex_, RELEASE writer_mutex_, +/// then call this. Destroying it while writer_mutex_ is still held would order +/// writer_mutex_ before plugin_mutex(), and open_bag_writer() takes them the other way +/// round, which is a deadlock cycle. +void destroy_writer_under_plugin_lock(std::unique_ptr & writer) { + if (!writer) { + return; + } + std::lock_guard plock(plugin_mutex()); + writer.reset(); +} + /// Bound the probe reason so a verbose pluginlib error does not flood the log. std::string truncate_reason(const std::string & reason, size_t max_len = 200) { if (reason.size() <= max_len) { @@ -212,6 +278,21 @@ RosbagCapture::RosbagCapture(rclcpp::Node * node, FaultStorage * storage, const RosbagCapture::~RosbagCapture() { stop(); + + // Whatever stop() did not close is closed here, explicitly, rather than left to + // implicit member destruction - which would run ~Writer under no lock at all and + // is the one destruction site in this class that cannot be found by looking for + // a reset(). It is reachable: stop() returns early when the capture was never + // started or is already stopped, and a confirmation that read running_ as true + // just before stop() cleared it can still install a writer after + // finalize_post_fault_recording() has been and gone. Normally a no-op. + std::unique_ptr closing_writer; + { + std::lock_guard wlock(writer_mutex_); + closing_writer = std::move(active_writer_); + created_topics_.clear(); + } + destroy_writer_under_plugin_lock(closing_writer); } void RosbagCapture::start() { @@ -437,17 +518,29 @@ void RosbagCapture::on_fault_confirmed(const std::string & fault_code) { double recording_span_sec = 0.0; { std::lock_guard wlock(writer_mutex_); - // Taken out rather than reset here, and measured before it is closed. Dropping - // the last reference runs ~Writer, which flushes the bag and writes - // metadata.yaml, and unique_ptr::reset() only returns once that is done - so - // resetting first would charge the close to the recording, which is exactly - // what the finalise path is careful not to do. + // Measured before the close below, not after: dropping the last + // reference runs ~Writer, which flushes the bag and writes + // metadata.yaml, and that only returns once the I/O is done - so + // measuring after would charge the close to the recording, which is + // exactly what the finalise path is careful not to do. closing_writer = std::move(active_writer_); created_topics_.clear(); recording_span_sec = span_sec_since(recording_started_at_ns_.exchange(0)); } - // Outside the lock, and before the size is measured: metadata.yaml has to exist. - closing_writer.reset(); + // Closed outside writer_mutex_ and under plugin_mutex() instead. Both halves + // of that matter. Keeping the close off writer_mutex_ was the original + // author's call and the reason still holds: flush + the metadata.yaml write + // are real I/O, and writer_mutex_ is what message_callback takes for every + // post-fault message, so a close held under it stalls the capture's own write + // path. What running the close unsynchronized cost instead was safety: + // ~Writer can unload the storage plugin's shared library while another Writer + // for the same format is loading it, anywhere in the process. plugin_mutex() + // orders exactly those two against each other and nothing else - see its + // definition for the crash and the measurements. The cost is paid by a + // concurrent open_bag_writer(), which now waits behind this close. + destroy_writer_under_plugin_lock(closing_writer); + // The size is measured after the close: metadata.yaml exists only once the + // writer is gone, and nothing here still needs either lock. size_t bag_size = calculate_bag_size(bag_path); @@ -954,11 +1047,19 @@ std::string RosbagCapture::get_topic_type(const std::string & topic) const { } void RosbagCapture::discard_active_writer(const std::string & bag_path) { + // Same two-step shape as every other destruction site: out of the member under + // writer_mutex_, then destroyed under plugin_mutex() with writer_mutex_ already + // released. Resetting in place under writer_mutex_ - which is how this read + // before - would take writer_mutex_ then plugin_mutex(), the reverse of + // open_bag_writer(), and close a deadlock cycle. + std::unique_ptr closing_writer; { std::lock_guard wlock(writer_mutex_); - active_writer_.reset(); + closing_writer = std::move(active_writer_); created_topics_.clear(); } + destroy_writer_under_plugin_lock(closing_writer); + std::error_code ec; std::filesystem::remove_all(bag_path, ec); } @@ -973,21 +1074,39 @@ std::optional RosbagCapture::open_bag_writer(const std::string & fa std::filesystem::create_directories(bag_dir.parent_path()); } - // Create writer and store as member for post-fault recording - std::lock_guard wlock(writer_mutex_); - active_writer_ = std::make_unique(); - created_topics_.clear(); - rosbag2_storage::StorageOptions storage_options; storage_options.uri = bag_path; storage_options.storage_id = config_.format; storage_options.max_bagfile_size = config_.max_bag_size_mb * 1024 * 1024; - active_writer_->open(storage_options); + // Construct AND open under plugin_mutex(): open() is the call that loads the + // storage plugin, so a close running concurrently anywhere in the process is + // what has to be excluded. The constructor is in the same critical section + // because a Writer that fails to open is destroyed on the way out of this + // scope - the local is declared after the guard, so unwinding destroys it + // first and the failed writer's destructor is still covered. + std::lock_guard plock(plugin_mutex()); + auto writer = std::make_unique(); + writer->open(storage_options); + + // Published into the member only once it is open, under writer_mutex_ and + // nested inside plugin_mutex(). This is the one place the two are held + // together, and this is the direction every other path must agree with: + // plugin_mutex() -> writer_mutex_, never the reverse. Taking writer_mutex_ + // here rather than around open() also keeps message_callback off the open + // I/O it used to wait behind. + std::lock_guard wlock(writer_mutex_); + active_writer_ = std::move(writer); + created_topics_.clear(); return bag_path; } catch (const std::exception & e) { RCLCPP_ERROR(node_->get_logger(), "Failed to open bag file '%s': %s", bag_path.c_str(), e.what()); + // Both guards above are gone by the time this runs (unwinding destroyed + // them), so discard_active_writer() is free to take them again in its own + // order. active_writer_ is untouched by a failed open - the writer only + // reaches the member once open() has returned - so this is here for the + // partial directory on disk. discard_active_writer(bag_path); return std::nullopt; } @@ -1358,11 +1477,22 @@ void RosbagCapture::finalize_post_fault_recording() { } } - // Close outside both locks. Flushing the bag and writing metadata.yaml is real - // I/O, and the writer is exclusively ours now, so nothing is gained by holding - // the post-roll state or the write path hostage while it finishes. It must, - // however, close BEFORE the size is measured below. - closing_writer.reset(); + // Closed outside all three of post_fault_timer_mutex_, capture_topics_mutex_ and + // writer_mutex_: flushing the bag and writing metadata.yaml is real I/O, and the + // writer is exclusively ours now (taken out above), so nothing is gained by + // holding the post-roll state or the write path hostage while it finishes. That + // reasoning was the original author's and it still holds - writer_mutex_ in + // particular is taken by message_callback for every post-fault message, and a + // close charged to it stalls the next recording's write path, not just its open. + // + // What it does NOT buy is safety, which is what plugin_mutex() is for: ~Writer + // can unload the storage plugin's shared library while another Writer for the + // same format is loading it, anywhere in the process, and that double unload + // segfaults in rcutils_unload_shared_library. The close is therefore charged to + // the plugin lock and to nothing else, and a concurrent open_bag_writer() waits + // behind it. It must also close BEFORE the size is measured below: metadata.yaml + // is written by the destructor. + destroy_writer_under_plugin_lock(closing_writer); // Calculate final size and store metadata size_t bag_size = calculate_bag_size(bag_path); @@ -1430,19 +1560,39 @@ std::optional RosbagCapture::default_storage_probe(const std::strin // Probe the plugin by opening a throwaway bag. Returns the failure reason (never // throws) when the plugin is missing or unusable, so the caller can fall back or // self-disable instead of crashing. - const std::string test_path = - std::filesystem::temp_directory_path().string() + "/.rosbag_format_test_" + std::to_string(getpid()); + // Unique per CALL, not per process. Two captures being constructed at the same + // time - a second FaultManagerNode, or one created while another records - each + // probe both backends, and a path keyed only by pid put them in the same + // directory: one probe's remove_all() below then deletes the bag the other is + // still writing. rosbag2 writes metadata.yaml from ~Writer, so that lands as an + // exception escaping a destructor rather than as a probe failure this function + // could report. plugin_mutex() serialises the open and the close, but the + // remove_all() is deliberately outside it, so the paths have to differ. + static std::atomic probe_seq{0}; + const std::string test_path = std::filesystem::temp_directory_path().string() + "/.rosbag_format_test_" + + std::to_string(getpid()) + "_" + + std::to_string(probe_seq.fetch_add(1, std::memory_order_relaxed)); std::optional error; - try { - rosbag2_cpp::Writer writer; - rosbag2_storage::StorageOptions opts; - opts.uri = test_path; - opts.storage_id = format; - writer.open(opts); - } catch (const std::exception & e) { - // Keep the reason (plugin missing vs. I/O / permission) for the caller's log. - error = e.what(); + { + // The probe loads and unloads a storage plugin exactly like a real recording + // does, so it takes the same lock. It ran unsynchronized before, which left the + // whole hazard open on the one path most likely to hit it: a fault manager + // constructed while another one is recording probes both formats back to back, + // and construction is precisely when a second capture appears in the process. + // The guard is outside the try so the throwaway writer, declared inside, is + // destroyed under it on the failure path too. + std::lock_guard plock(plugin_mutex()); + try { + rosbag2_cpp::Writer writer; + rosbag2_storage::StorageOptions opts; + opts.uri = test_path; + opts.storage_id = format; + writer.open(opts); + } catch (const std::exception & e) { + // Keep the reason (plugin missing vs. I/O / permission) for the caller's log. + error = e.what(); + } } std::error_code ec; diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp index d7b9d3f66..b3baefee6 100644 --- a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp @@ -16,6 +16,8 @@ #include #include +#include +#include #include #include #include @@ -40,6 +42,45 @@ using ros2_medkit_fault_manager::RosbagCapture; using ros2_medkit_fault_manager::RosbagConfig; using ros2_medkit_fault_manager::SnapshotConfig; +namespace { + +/// Multiplier for the wall-clock window and throughput thresholds +/// ConcurrentCapturesInOneProcessSurviveEachOthersPluginTraffic asserts. The +/// sanitizer CI jobs run this suite too, and an ASan/TSan-instrumented open/ +/// close/dlclose cycle does materially less work per wall-clock second, so a +/// window or a throughput floor that is tight unsanitized can go unmet under +/// instrumentation for a reason that has nothing to do with the plugin-loader +/// race the test exists to catch. Mirrors test_cancel_outcomes.cpp's reader in +/// ros2_medkit_gateway (same variable, same jobs, same semantics): the +/// sanitizer jobs export MEDKIT_TEST_TIME_SCALE with the same factor they +/// apply to every ctest TIMEOUT; unset / unparseable / below 1 means no +/// scaling, so the normal job keeps the tight window and thresholds. +double test_time_scale() { + const char * raw = std::getenv("MEDKIT_TEST_TIME_SCALE"); + if (raw == nullptr) { + return 1.0; + } + try { + const double scale = std::stod(raw); + return scale >= 1.0 ? scale : 1.0; + } catch (const std::exception &) { + return 1.0; + } +} + +/// Scale a wall-clock budget by test_time_scale(). +std::chrono::milliseconds scaled(std::chrono::milliseconds base) { + return std::chrono::milliseconds{static_cast(static_cast(base.count()) * test_time_scale())}; +} + +/// Scale a throughput floor by test_time_scale(), so a longer scaled window still +/// demands proportionally as much work, not just a longer wait for the same count. +int scaled(int base) { + return static_cast(static_cast(base) * test_time_scale()); +} + +} // namespace + class RosbagCaptureTest : public ::testing::Test { protected: void SetUp() override { @@ -2214,3 +2255,116 @@ TEST_F(RosbagCaptureIntegrationTest, ReconfirmingTheRecordingsOwnFaultResolvesNo capture.stop(); } + +TEST_F(RosbagCaptureIntegrationTest, ConcurrentCapturesInOneProcessSurviveEachOthersPluginTraffic) { + // The storage-plugin loader's state is PROCESS-GLOBAL: class_loader keys one + // registry of loaded libraries by library path, so every writer of a format shares + // one shared-library handle no matter which capture opened it. A lock that is an + // instance member cannot order two captures against each other, and the failure is + // a double dlclose - the winner zeroes the handle, the loser calls a null + // deallocate and the process dies inside rcutils_unload_shared_library. + // + // Three threads therefore run at once, in the production shape. Two construct and + // destroy captures, whose constructors probe a backend by opening and closing a + // throwaway bag. One stands in for the capture pool and confirms faults, which + // OPENS bags. This thread spins, so it is the executor: the post-fault timer fires + // here and CLOSES them. Every pairing of an open against a close is crossed, across + // instance boundaries, which is the part no single-instance test reaches. + // + // Two claims are falsifiable here, and both were checked by mutation. + // + // Make the plugin lock per-thread rather than process-wide: this segfaults, 3 runs + // of 3. + // + // Close the finalise's writer under writer_mutex_ instead, so that lock is held + // across the plugin lock while open_bag_writer() takes them the other way round: + // this deadlocks, not crashes, on the first run. The confirming thread holds the + // plugin lock waiting for writer_mutex_ while the finalise holds writer_mutex_ + // waiting for the plugin lock. A bare SIGTERM does not end it, because rclcpp's + // handler runs into the same deadlock - but ctest is not fooled by that: it reaps + // the hung process at this test's own TIMEOUT and reports the failure by name, + // which is what CI actually shows for this variant, not an indefinitely hung run. + // + // Both halves need the confirmations to come off THIS thread; one thread cannot + // deadlock against itself. The cycle also needs the close to sit where the design + // puts it, OUTSIDE post_fault_timer_mutex_: a close moved back inside that lock + // cannot deadlock, because a confirmation blocks on post_fault_timer_mutex_ in + // attach_to_active_recording() before it ever reaches the plugin lock - it just + // stalls every confirmation for the length of a bag close, which is the cost the + // design declines to pay. That variant passes, 3 runs of 3, so it is not what this + // test pins. + auto rosbag_config = create_rosbag_config(); + rosbag_config.duration_sec = 2.0; + // Short windows on purpose: the post-fault timer then fires often, so this thread + // is closing bags for most of the run rather than a handful of times. + rosbag_config.duration_after_sec = 0.1; + auto snapshot_config = create_snapshot_config(); + + // lazy_start, so the probing captures never create subscriptions on the shared + // node. Concurrent create_generic_subscription() on ONE node is a different race + // (the rcutils_hash_map one, serialised per instance by node_ops_mutex_), and + // letting it fire here would make a crash unattributable. + auto probe_config = create_rosbag_config(); + probe_config.lazy_start = true; + + RosbagCapture capture(node_.get(), storage_.get(), rosbag_config, snapshot_config); + auto pub = node_->create_publisher("/plugin_race_probe", 10); + capture.start(); + publish_for(pub, std::chrono::milliseconds(600)); + + std::atomic stop_racers{false}; + std::atomic probes_done{0}; + auto probe_loop = [&]() { + while (!stop_racers.load()) { + RosbagCapture probe(node_.get(), storage_.get(), probe_config, snapshot_config); + probes_done.fetch_add(1); + } + }; + std::thread racer_a(probe_loop); + std::thread racer_b(probe_loop); + + // The capture pool's stand-in. Each confirmation either opens a bag or attaches to + // the window this thread's timer is about to close; only this thread ever confirms, + // which is the contract the production caller keeps via the node rosbag mutex. + std::atomic confirms_done{0}; + std::thread confirmer([&]() { + int n = 0; + while (!stop_racers.load()) { + capture.on_fault_confirmed("PLUGIN_RACE_" + std::to_string(n++)); + confirms_done.fetch_add(1); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + }); + + // This thread is the executor for the whole run: spinning is what fires the + // post-fault timer, so every recording the confirmer opens is closed here. + // Scaled by test_time_scale(): an instrumented open/close/dlclose cycle does + // less work per wall-clock second, so the window has to grow to still cross + // the same amount of traffic under a sanitizer job. + publish_for(pub, scaled(std::chrono::milliseconds(5000))); + + stop_racers.store(true); + racer_a.join(); + racer_b.join(); + confirmer.join(); + + // A crash or a hang is the real assertion; these say the run did the work it + // claims. Traffic that never got going would leave the opens and closes uncrossed + // and prove nothing, and rows are what say the recordings really finalised while + // the loader was under load. Thresholds scaled the same way as the window above, + // so a longer scaled window still demands proportionally as much work rather than + // just a longer wait for the same absolute count. + EXPECT_GT(probes_done.load(), scaled(50)) << "the probing threads barely ran, so no open/close traffic was crossed"; + EXPECT_GT(confirms_done.load(), scaled(50)) + << "the confirming thread barely ran, so few bags were opened off-executor"; + size_t rows = 0; + for (int i = 0; i < confirms_done.load(); ++i) { + if (storage_->get_rosbag_file("PLUGIN_RACE_" + std::to_string(i)).has_value()) { + ++rows; + } + } + EXPECT_GT(rows, static_cast(scaled(10))) + << "recordings did not finalise while the plugin loader was under load"; + + capture.stop(); +} diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_integration.test.py b/src/ros2_medkit_fault_manager/test/test_rosbag_integration.test.py index 551e42094..1ee2d1df5 100644 --- a/src/ros2_medkit_fault_manager/test/test_rosbag_integration.test.py +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_integration.test.py @@ -28,7 +28,7 @@ import rclpy from rclpy.node import Node from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy -from ros2_medkit_msgs.msg import Fault +from ros2_medkit_msgs.msg import Fault, FaultEvent from ros2_medkit_msgs.srv import ClearFault, GetRosbag, ReportFault from sensor_msgs.msg import Temperature @@ -284,11 +284,18 @@ def _wait_for_buffered_data(self, count=5, timeout=8.0): Wait until the rosbag ring buffer holds fresh data before a confirmation. A previous fault's post-fault recording window diverts incoming messages - away from the ring buffer, so right after it closes the buffer can be - empty and flush_to_bag creates no bag. There is no service to read the - buffer, so subscribe to the same source the fault manager records and - wait until ``count`` messages arrive - by then the co-subscribed fault - manager has buffered a comparable amount. Adaptive, unlike a fixed sleep. + away from the ring buffer - message_callback() writes straight to the open + bag and returns before it ever touches the buffer - so right after the + window closes the buffer can be empty and flush_to_bag creates no bag. + There is no service to read the buffer, so this counts messages on the same + source the fault manager records instead, as a proxy. Counting alone is not + a safe proxy: a confirmation landing mid-count reopens that same diversion, + so ``count`` messages arriving at this node can still correspond to + near-zero buffered history on the fault manager's side. The broad topic + modes already subscribe to ``/fault_manager/events``, so this does too, and + resets the count on every ``EVENT_CONFIRMED`` - guaranteeing the streak of + messages this function finally returns on arrived with no post-fault window + newly opened partway through it. """ received = 0 @@ -296,19 +303,34 @@ def _cb(_msg): nonlocal received received += 1 + def _on_event(msg): + nonlocal received + if msg.event_type == FaultEvent.EVENT_CONFIRMED: + received = 0 + # Match the publisher's BEST_EFFORT sensor QoS, else no messages arrive. qos = QoSProfile( reliability=ReliabilityPolicy.BEST_EFFORT, history=HistoryPolicy.KEEP_LAST, depth=10, ) + # Matches the fault manager's own reliable, depth-100 events publisher. + events_qos = QoSProfile( + reliability=ReliabilityPolicy.RELIABLE, + history=HistoryPolicy.KEEP_LAST, + depth=100, + ) sub = self.node.create_subscription(Temperature, '/test/temperature', _cb, qos) + events_sub = self.node.create_subscription( + FaultEvent, '/fault_manager/events', _on_event, events_qos + ) try: deadline = time.time() + timeout while received < count and time.time() < deadline: rclpy.spin_once(self.node, timeout_sec=0.1) finally: self.node.destroy_subscription(sub) + self.node.destroy_subscription(events_sub) return received >= count def test_01_rosbag_created_on_fault_confirmation(self): From ee73ba866994a1e6cfb80c0497f72a4d5d64c429 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Thu, 13 Aug 2026 12:29:15 +0200 Subject: [PATCH 3/4] fix(fault_manager): name the apt package in every storage warning The warning that fires when neither storage backend loads named the ROS package names, rosbag2_storage_mcap and rosbag2_storage_default_plugins. Those are not what an operator installs: the installable names carry the distro prefix and use hyphens, so the reader is left to work out the translation at the moment black-box capture has just disabled itself. The fallback warning next to it already printed the installable name, so the two disagreed on what a package name is. Both now go through one helper that returns the installable name, and the disable warning prints both names as a command line that can be pasted. The config reference listed rosbag.format twice with contradictory defaults, sqlite3 in one row and mcap in the other. The mcap row is the correct one; the stale row is removed. The fault manager README called rosbag2_storage_default_plugins "the sqlite3 plugin", which holds on humble but not on jazzy or lyrical, where it is a metapackage that pulls in mcap as well. A test captures the rcutils output around the constructor for all three storage outcomes and requires the installable name in each message. --- docs/config/fault-manager.rst | 4 - src/ros2_medkit_fault_manager/README.md | 4 +- .../src/rosbag_capture.cpp | 35 +++-- .../test/test_rosbag_capture.cpp | 128 ++++++++++++++++++ 4 files changed, 156 insertions(+), 15 deletions(-) diff --git a/docs/config/fault-manager.rst b/docs/config/fault-manager.rst index 6ef4991e4..3b425418a 100644 --- a/docs/config/fault-manager.rst +++ b/docs/config/fault-manager.rst @@ -293,10 +293,6 @@ Capture continuous rosbag recordings around fault events. * - ``rosbag.exclude_topics`` - ``[]`` - Topics to drop from whatever the selection mode picked. - * - ``rosbag.format`` - - ``sqlite3`` - - Storage format handed to rosbag2. Use ``mcap`` where the reader expects - it. * - ``rosbag.storage_path`` - ``""`` - Directory the bag files are written to. Empty falls back to the system diff --git a/src/ros2_medkit_fault_manager/README.md b/src/ros2_medkit_fault_manager/README.md index b325273fc..83cdd12af 100644 --- a/src/ros2_medkit_fault_manager/README.md +++ b/src/ros2_medkit_fault_manager/README.md @@ -118,8 +118,8 @@ unknown format string lands on. Neither backend is privileged over the other: whichever is configured, if its plugin is unavailable at startup the FaultManager logs a warning naming the missing package and falls back to the other one automatically, disabling capture only if neither loads. Both `rosbag2_storage_mcap` -and `rosbag2_storage_default_plugins` (the sqlite3 plugin) are runtime dependencies -of this package, so a normal install pulls in both backends. See +and `rosbag2_storage_default_plugins` (which carries the sqlite3 plugin) are runtime +dependencies of this package, so a normal install pulls in both backends. See `docs/config/fault-manager.rst` and `docs/tutorials/snapshots.rst` for the full parameter list and recording lifecycle. diff --git a/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp b/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp index b5d2f0bbd..d83fa5ac7 100644 --- a/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/src/rosbag_capture.cpp @@ -41,17 +41,31 @@ namespace ros2_medkit_fault_manager { namespace { -/// Actionable install hint for an unavailable storage backend. -std::string storage_plugin_hint(const std::string & format) { +/// Installable package name for the storage plugin backing `format`, empty for +/// any other format. +/// +/// The name an operator has to type, not the ROS package name: those differ by +/// the distro prefix and by underscore-versus-hyphen, so a message carrying the +/// ROS name leaves the reader to guess the translation. +std::string storage_plugin_package(const std::string & format) { const char * distro = std::getenv("ROS_DISTRO"); const std::string d = (distro && *distro) ? distro : "$ROS_DISTRO"; if (format == "mcap") { - return "install ros-" + d + "-rosbag2-storage-mcap"; + return "ros-" + d + "-rosbag2-storage-mcap"; } if (format == "sqlite3") { - return "install ros-" + d + "-rosbag2-storage-default-plugins"; + return "ros-" + d + "-rosbag2-storage-default-plugins"; + } + return {}; +} + +/// Actionable install hint for an unavailable storage backend. +std::string storage_plugin_hint(const std::string & format) { + const std::string package = storage_plugin_package(format); + if (package.empty()) { + return "check the rosbag2 storage plugin installation"; } - return "check the rosbag2 storage plugin installation"; + return "install " + package; } /// Monotonic nanoseconds, for measuring how long something took. @@ -247,11 +261,14 @@ RosbagCapture::RosbagCapture(rclcpp::Node * node, FaultStorage * storage, const const std::string reason = truncate_reason(*probe_err); const std::string other_format = (config_.format == "sqlite3") ? "mcap" : "sqlite3"; if (auto other_err = storage_probe_(other_format)) { + // Both names are known here - the check above left `config_.format` as one + // of the two formats - so this is a command line an operator can paste. + const std::string packages = storage_plugin_package(config_.format) + " " + storage_plugin_package(other_format); RCLCPP_WARN(node_->get_logger(), - "No usable rosbag storage backend: '%s' (%s) and '%s' (%s) both failed to load; install " - "rosbag2_storage_mcap and rosbag2_storage_default_plugins. Black-box rosbag capture disabled " - "(freeze-frame snapshots still work)", - config_.format.c_str(), reason.c_str(), other_format.c_str(), truncate_reason(*other_err).c_str()); + "No usable rosbag storage backend: '%s' (%s) and '%s' (%s) both failed to load; install %s. " + "Black-box rosbag capture disabled (freeze-frame snapshots still work)", + config_.format.c_str(), reason.c_str(), other_format.c_str(), truncate_reason(*other_err).c_str(), + packages.c_str()); config_.enabled = false; return; } diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp index b3baefee6..1dbe5afb3 100644 --- a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp @@ -14,14 +14,18 @@ #include +#include #include #include +#include #include +#include #include #include #include #include #include +#include #include #include #include @@ -44,6 +48,80 @@ using ros2_medkit_fault_manager::SnapshotConfig; namespace { +/// Captures rcutils log output while alive and restores the console handler on +/// every exit path - leaving the process-global handler installed would swallow +/// the output of every later case in this binary. +/// +/// The handler is a plain C function pointer with no user-data slot, so the live +/// capture is reached through a file-static. Only the test thread logs in the +/// cases that use this, but the pointer is atomic because the handler is +/// process-global and other threads in this binary do log. +class LogCapture { + public: + LogCapture() { + active().store(this); + rcutils_logging_set_output_handler(&LogCapture::handler); + } + ~LogCapture() { + rcutils_logging_set_output_handler(rcutils_logging_console_output_handler); + active().store(nullptr); + } + LogCapture(const LogCapture &) = delete; + LogCapture & operator=(const LogCapture &) = delete; + LogCapture(LogCapture &&) = delete; + LogCapture & operator=(LogCapture &&) = delete; + + /// How many captured lines contain `needle`. + int count(const std::string & needle) const { + std::lock_guard lk(mutex_); + return static_cast(std::count_if(lines_.begin(), lines_.end(), [&needle](const std::string & line) { + return line.find(needle) != std::string::npos; + })); + } + + private: + static std::atomic & active() { + static std::atomic current{nullptr}; + return current; + } + + static void handler(const rcutils_log_location_t * /*location*/, int /*severity*/, const char * /*name*/, + rcutils_time_point_value_t /*timestamp*/, const char * format, va_list * args) { + char buf[1024]; + va_list copy; + va_copy(copy, *args); + // The format string arrives from the logging call site through the handler + // signature, so there is no literal to write here. The build runs + // -Werror=format=2; GCC exempts va_list-taking formatters from + // -Wformat-nonliteral, clang does not, and clang-tidy does not honour + // suppression comments for a diagnostic raised as an error. Scoped to the + // single call. +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wformat-nonliteral" + vsnprintf(buf, sizeof(buf), format, copy); +#pragma GCC diagnostic pop + va_end(copy); + LogCapture * capture = active().load(); + if (capture == nullptr) { + return; + } + std::lock_guard lk(capture->mutex_); + capture->lines_.emplace_back(buf); + } + + mutable std::mutex mutex_; + std::vector lines_; +}; + +/// The apt package name the storage warnings must name for `format`, built the +/// same way the code under test builds it so the assertion does not have to +/// hardcode a distro. +std::string expected_storage_package(const std::string & format) { + const char * distro = std::getenv("ROS_DISTRO"); + const std::string d = (distro != nullptr && *distro != '\0') ? distro : "$ROS_DISTRO"; + return (format == "mcap") ? "ros-" + d + "-rosbag2-storage-mcap" : "ros-" + d + "-rosbag2-storage-default-plugins"; +} + /// Multiplier for the wall-clock window and throughput thresholds /// ConcurrentCapturesInOneProcessSurviveEachOthersPluginTraffic asserts. The /// sanitizer CI jobs run this suite too, and an ASan/TSan-instrumented open/ @@ -256,6 +334,56 @@ TEST_F(RosbagCaptureTest, NoUsableBackendDisablesCaptureWithoutCrashing) { EXPECT_FALSE(rb->is_enabled()); } +TEST_F(RosbagCaptureTest, StorageWarningsNameInstallablePackages) { + // Both warnings that report an unavailable backend have to name what an + // operator installs, which is the apt package - not the ROS package name, + // which differs from it by the distro prefix and by underscore-versus-hyphen + // and leaves the reader to work out the translation. + auto snapshot_config = create_snapshot_config(); + const std::string mcap_package = expected_storage_package("mcap"); + const std::string sqlite_package = expected_storage_package("sqlite3"); + + // Neither backend loads: capture disables, and the message names both packages. + { + auto rosbag_config = create_rosbag_config(); + rosbag_config.format = "mcap"; + RosbagCapture::StorageProbeFn none_usable = [](const std::string &) -> std::optional { + return std::string("simulated: backend unavailable"); + }; + LogCapture logs; + RosbagCapture rb(node_.get(), storage_.get(), rosbag_config, snapshot_config, none_usable); + ASSERT_FALSE(rb.is_enabled()); + EXPECT_EQ(logs.count("install " + mcap_package + " " + sqlite_package), 1); + } + + // Only the configured backend fails: capture falls back, and the message names + // the package for the format the operator asked for and lost. + { + auto rosbag_config = create_rosbag_config(); + rosbag_config.format = "mcap"; + RosbagCapture::StorageProbeFn mcap_missing = [](const std::string & f) -> std::optional { + return (f == "mcap") ? std::optional("simulated: mcap plugin not found") : std::nullopt; + }; + LogCapture logs; + RosbagCapture rb(node_.get(), storage_.get(), rosbag_config, snapshot_config, mcap_missing); + ASSERT_TRUE(rb.is_enabled()); + EXPECT_EQ(logs.count("install " + mcap_package), 1); + } + + // The symmetric fallback, so neither package name can be hardcoded and pass. + { + auto rosbag_config = create_rosbag_config(); + rosbag_config.format = "sqlite3"; + RosbagCapture::StorageProbeFn sqlite_missing = [](const std::string & f) -> std::optional { + return (f == "sqlite3") ? std::optional("simulated: sqlite3 plugin not found") : std::nullopt; + }; + LogCapture logs; + RosbagCapture rb(node_.get(), storage_.get(), rosbag_config, snapshot_config, sqlite_missing); + ASSERT_TRUE(rb.is_enabled()); + EXPECT_EQ(logs.count("install " + sqlite_package), 1); + } +} + // @verifies REQ_INTEROP_088 TEST_F(RosbagCaptureTest, Sqlite3BaselineUnavailableDisablesCapture) { // When sqlite3 is configured and neither it nor its mcap fallback loads, From d339011362c3813fcf25754687588da54d0cc827 Mon Sep 17 00:00:00 2001 From: Bartosz Burda Date: Thu, 13 Aug 2026 18:12:03 +0200 Subject: [PATCH 4/4] test(fault_manager): make the unknown-format case prove normalisation ConstructorFallsBackOnUnknownFormat ran against the real probe and accepted either backend, so it could not fail. Delete the normalisation and an unknown format still resolves to sqlite3: its own probe fails, and because the format is not literally "sqlite3" the symmetric fallback picks sqlite3. Both paths end on the same value, so the assertion held either way and the documented rule - an unknown format string falls back to sqlite3 - had no coverage. The case now injects a probe that reports every backend usable, which leaves normalisation as the only thing that can change the format, and asserts the value exactly. Deleting the normalisation fails it. The reason the assertion had been relaxed no longer applies: it allowed for a host carrying mcap but not sqlite3, and both plugins are test dependencies of this package now, as the parametrised bag tests in the same binary already require. Also declare rcutils/logging.h, which the log capture uses directly and was reaching only through rclcpp.hpp, and drop an unused launch_testing.markers import. launch_testing exports asserts from its own __init__, so nothing in that file depended on the removed import. --- .../test/test_rosbag_capture.cpp | 31 ++++++++++--------- .../test/test_rosbag_default_format.test.py | 1 - 2 files changed, 17 insertions(+), 15 deletions(-) diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp index 1dbe5afb3..5c79fa5a6 100644 --- a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp @@ -35,6 +35,7 @@ #include #include "rclcpp/rclcpp.hpp" +#include "rcutils/logging.h" #include "ros2_medkit_fault_manager/fault_storage.hpp" #include "ros2_medkit_fault_manager/rosbag_capture.hpp" #include "ros2_medkit_fault_manager/snapshot_capture.hpp" @@ -248,26 +249,28 @@ TEST_F(RosbagCaptureTest, DefaultFormatIsMcap) { // @verifies REQ_INTEROP_088 TEST_F(RosbagCaptureTest, ConstructorFallsBackOnUnknownFormat) { - // An unknown format string must NOT terminate the node. It normalises to - // "sqlite3" first, then goes through the same real probe as an explicit - // sqlite3 configuration would, so this test uses the real probe on - // purpose - unlike the crash-safety tests below, which inject a double. - // That means the resolved format is not fixed: on a host with only mcap - // installed and not sqlite3 (the buildfarm chroot before - // rosbag2_storage_default_plugins lands as a real dependency), resolution - // falls back to mcap instead, which is the fallback being symmetric doing - // exactly its job, not a failure. The contract this test is actually about - // is that capture stays enabled and resolves to one of the two real - // backends, not "invalid_format" - not which one that happens to be here. + // An unknown format string must NOT terminate the node: it normalises to + // "sqlite3" before any probe runs. The probe is injected and reports every + // backend usable, so no fallback can move the format afterwards and the + // assertion pins normalisation and nothing else. + // + // The real probe cannot be used here. An unknown format reaches "sqlite3" + // down a second path with it - its own probe fails, and because the format + // is not literally "sqlite3" the fallback picks sqlite3 - so the two paths + // are indistinguishable by the resolved value and the assertion would hold + // whether normalisation ran or not. auto rosbag_config = create_rosbag_config(); rosbag_config.format = "invalid_format"; auto snapshot_config = create_snapshot_config(); + RosbagCapture::StorageProbeFn all_usable = [](const std::string &) -> std::optional { + return std::nullopt; + }; std::shared_ptr rb; - EXPECT_NO_THROW(rb = std::make_shared(node_.get(), storage_.get(), rosbag_config, snapshot_config)); + EXPECT_NO_THROW( + rb = std::make_shared(node_.get(), storage_.get(), rosbag_config, snapshot_config, all_usable)); ASSERT_NE(rb, nullptr); EXPECT_TRUE(rb->is_enabled()); - EXPECT_TRUE(rb->config().format == "sqlite3" || rb->config().format == "mcap") - << "resolved format was '" << rb->config().format << "', neither a real backend"; + EXPECT_EQ(rb->config().format, "sqlite3"); } // The crash-safety branches below force the storage probe via an injected double, diff --git a/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py b/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py index 4946eded8..a18b047f3 100644 --- a/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py @@ -30,7 +30,6 @@ import launch.actions import launch_ros.actions import launch_testing.actions -import launch_testing.markers import rclpy from rclpy.node import Node from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy