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..3b425418a 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 @@ -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 @@ -306,6 +302,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 +388,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..83cdd12af 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` (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. + ### 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/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/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..d83fa5ac7 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 @@ -37,14 +41,31 @@ namespace ros2_medkit_fault_manager { namespace { +/// 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 "ros-" + d + "-rosbag2-storage-mcap"; + } + if (format == "sqlite3") { + 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) { - 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"; + 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. @@ -80,6 +101,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) { @@ -163,9 +246,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 +259,28 @@ 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. + 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(), - "sqlite3 rosbag storage is unavailable (%s); the rosbag2 base install may be broken. " + "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)", - reason.c_str()); - config_.enabled = false; - return; - } - if (auto sqlite_err = storage_probe_("sqlite3")) { - 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()); + config_.format.c_str(), reason.c_str(), other_format.c_str(), truncate_reason(*other_err).c_str(), + packages.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)", @@ -212,6 +295,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 +535,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 +1064,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 +1091,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 +1494,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 +1577,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 c88b67f8c..5c79fa5a6 100644 --- a/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_capture.cpp @@ -14,12 +14,18 @@ #include +#include #include #include +#include +#include +#include +#include #include #include #include #include +#include #include #include #include @@ -29,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" @@ -40,6 +47,119 @@ using ros2_medkit_fault_manager::RosbagCapture; using ros2_medkit_fault_manager::RosbagConfig; 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/ +/// 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 { @@ -118,15 +238,36 @@ 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" 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_EQ(rb->config().format, "sqlite3"); @@ -156,6 +297,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 @@ -173,10 +337,62 @@ 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 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(); @@ -2170,3 +2386,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_default_format.test.py b/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py new file mode 100644 index 000000000..a18b047f3 --- /dev/null +++ b/src/ros2_medkit_fault_manager/test/test_rosbag_default_format.test.py @@ -0,0 +1,361 @@ +#!/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 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_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): 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;