diff --git a/docs/config/server.rst b/docs/config/server.rst index 0171b268a..bc8e79e84 100644 --- a/docs/config/server.rst +++ b/docs/config/server.rst @@ -384,6 +384,23 @@ Configure how the gateway connects to the fault manager services and event topic values, marked ``connected: false`` in the snapshot's ``x-medkit`` block. Explicit snapshot config in the fault manager always wins when present. Only active when plugins are loaded. + * - ``entity_freeze_frame.storage.path`` + - string + - ``""`` + - SQLite file the captured frames are persisted in, so a restart serves the + values frozen at fault time instead of re-reading the plant. When empty, + the frames go in ``entity_freeze_frames.db`` next to + ``triggers.storage.path``. With that empty too they stay in memory and are + lost on restart. A reloaded frame keeps its original ``captured_at`` and + its ``capture_origin``, and the startup catch-up then runs only for faults + that have no stored frame. The retained-frame bound of 256 faults counts + reloaded and freshly captured frames together, dropping the oldest first, + and a frame whose fault the fault manager no longer holds at all is + dropped at startup. A frame belonging to an occurrence that has since + been cleared and confirmed again is re-read at startup and marked + ``capture_origin: startup`` rather than served as the current one. When + that re-read cannot answer, the stale frame is discarded with a warning + naming the fault code and the entity. When ``fault_manager.namespace`` is set, the gateway also subscribes to the matching fault event topic (for example ``/robot1/fault_manager/events`` instead of the default diff --git a/docs/tutorials/snapshots.rst b/docs/tutorials/snapshots.rst index c23e845ef..2f7ef6b2f 100644 --- a/docs/tutorials/snapshots.rst +++ b/docs/tutorials/snapshots.rst @@ -201,15 +201,37 @@ and may predate the confirmation by the length of the outage; the entry's payload includes one, ``source_timestamp`` (the payload's own timestamp) alongside ``captured_at``. +With ``entity_freeze_frame.storage.path`` set, a captured frame survives a +gateway restart: it is reloaded at start and served exactly as it was +captured, with its original ``captured_at`` and no ``capture_origin`` marker. +Set the path to a file on a volume that outlives the container, or leave it +empty and the frames go next to the trigger store +(``triggers.storage.path``). With neither set they are process memory only +and a restart loses them. A plugin entity keeps exactly one frame per fault: a +re-confirm re-samples the plugin and replaces it, on disk as in memory, and a +re-confirm the gateway was down for is re-read at startup and marked +``capture_origin: startup``, so a confirmed occurrence never serves the +previous one's values. The startup comparison is against the fault's +``first_occurred``, which the fault manager resets on reactivation from +``CLEARED``, so it holds for exactly the faults the catch-up sees: a fault +that re-failed but has not re-confirmed yet is not in the confirmed list and +keeps its frame until it does confirm, and a ``HEALED`` to ``FAILED`` cycle +does not reset ``first_occurred`` at all (healing is off by default). If the +re-read cannot answer, because the entity is unreachable or serves nothing +usable, the stale frame is discarded rather than served, and the gateway warns +with the fault code and the entity so the missing evidence is not silent. + Faults that are already confirmed when the gateway starts are caught up at startup: the gateway lists the confirmed faults and captures a frame for each -plugin-backed one, so a device standing in fault across a gateway restart -still gets a frame. Catch-up frames carry ``"capture_origin": "startup"`` in -their ``x-medkit`` block because their values were read at gateway start, not -when the fault confirmed (which may be long before, since the fault manager +plugin-backed one that does not already have a stored frame, so a device +standing in fault across a gateway restart still gets a frame, and one whose +frame was already taken keeps the values from its own confirm edge instead of +today's. Catch-up frames carry ``"capture_origin": "startup"`` in their +``x-medkit`` block because their values were read at gateway start, not when +the fault confirmed (which may be long before, since the fault manager persists faults); ``captured_at`` always stamps the moment the values were -read. Frames without the marker were captured on the confirm edge. Disable -with: +read. Frames without the marker were captured on the confirm edge, and a +reloaded frame keeps whichever marker it was captured with. Disable with: .. code-block:: bash diff --git a/src/ros2_medkit_gateway/CHANGELOG.rst b/src/ros2_medkit_gateway/CHANGELOG.rst index 6c4dd39ee..4c802ae73 100644 --- a/src/ros2_medkit_gateway/CHANGELOG.rst +++ b/src/ros2_medkit_gateway/CHANGELOG.rst @@ -2,6 +2,11 @@ Changelog for package ros2_medkit_gateway ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +Forthcoming +----------- +* Entity freeze-frames survive a restart when ``entity_freeze_frame.storage.path`` names a database. Without one the frames lived in process memory, so a restart threw them away and the startup catch-up re-read the plant as it is now, serving today's values under the original fault and marking them ``x-medkit.capture_origin: startup`` - the values at fault time, which are the point of a freeze-frame, were gone. A reloaded frame is served exactly as it was captured: its original ``captured_at``, its own ``capture_origin`` (absent on a confirm-edge frame), and the ``connected`` / ``source_timestamp`` provenance it carried, and the startup catch-up then re-reads only the faults that have no frame. Leaving the path empty puts the store in ``entity_freeze_frames.db`` next to ``triggers.storage.path``. With neither set the frames stay in memory as before, and a store that cannot be opened or written is reported while the capture keeps working. The retained-frame bound of 256 faults counts reloaded and freshly captured frames together, a frame whose fault the fault_manager no longer holds in any status is dropped at startup (one reported as cleared keeps its frame), and a frame belonging to an occurrence that has since been cleared and re-confirmed is re-read at startup and marked ``capture_origin: startup`` rather than served as the current one +* Contributors: @bburda + 0.7.0 (2026-08-27) ------------------ * Rosbag bulk-data is addressed by recording id instead of fault code, so a fault holding several recordings can expose each one. ``GET /{entity}/bulk-data/rosbags`` now emits one descriptor per recording rather than one per fault - a burst that shares a bag used to appear as several entries each reporting the full bag size - and the covered faults move into ``x-medkit.fault_codes`` (was the scalar ``x-medkit.fault_code``). Old URLs keep working: an id that is not a recording is resolved as a fault code and serves that fault's newest recording, which is what it returned before. Authorization is unchanged in effect - a download is allowed when any fault the recording covers is in the entity's source scope, which is exactly the set that could reach it previously (`#623 `_, `#620 `_) diff --git a/src/ros2_medkit_gateway/CMakeLists.txt b/src/ros2_medkit_gateway/CMakeLists.txt index 23b20ac4d..527921338 100644 --- a/src/ros2_medkit_gateway/CMakeLists.txt +++ b/src/ros2_medkit_gateway/CMakeLists.txt @@ -840,6 +840,11 @@ if(BUILD_TESTING) target_link_libraries(test_entity_freeze_frame_capture gateway_ros2) medkit_target_dependencies(test_entity_freeze_frame_capture rclcpp ros2_medkit_msgs) + # Entity freeze-frame persistence (both store backends). Links gateway_core + # only (ROS-neutral). + medkit_add_gtest(test_entity_freeze_frame_store test/test_entity_freeze_frame_store.cpp) + target_link_libraries(test_entity_freeze_frame_store gateway_core) + # Add update manager tests medkit_add_gtest(test_update_manager test/test_update_manager.cpp) target_link_libraries(test_update_manager gateway_ros2) diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/entity_freeze_frame_store.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/entity_freeze_frame_store.hpp new file mode 100644 index 000000000..e76c0d08c --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/entity_freeze_frame_store.hpp @@ -0,0 +1,120 @@ +// 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. + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace ros2_medkit_gateway { + +/// One persisted entity freeze-frame: the row shape of the store, keyed by +/// (fault_code, entity_id). +/// +/// `frame` holds everything that is not already a column of its own - the +/// compact {resource_id: value} dict under "values", plus the payload +/// provenance the capture recorded ("connected", "source_timestamp") when the +/// plugin reported it. Keeping those in the blob rather than in columns is +/// what lets a reloaded frame be served byte for byte as it was captured. +struct StoredEntityFreezeFrame { + std::string fault_code; + std::string entity_id; + nlohmann::json frame; ///< {"values": {...}, "connected"?: bool, "source_timestamp"?: any} + int64_t captured_at_ns{0}; + std::string source; ///< capture path that read the values + std::string capture_origin; ///< "startup" for a catch-up frame, empty on a confirm edge +}; + +/// Persistence for the gateway's entity freeze-frames. +/// +/// The gateway's frames are process memory, so a restart re-derives them from +/// whatever the plant reads *now* - the values at fault time are gone and the +/// re-read is stamped with the restart. This store is what makes the captured +/// frame outlive the process. +/// +/// Writes are per fault code and wholesale: a re-confirm replaces every row +/// for that code, mirroring the in-memory map, whose entry for a code is +/// likewise replaced as a unit. Implementations must be thread-safe. +class EntityFreezeFrameStore { + public: + virtual ~EntityFreezeFrameStore() = default; + + /// Replace every row for @p fault_code with @p frames (one row per entity). + /// An empty vector leaves no rows for the code. + virtual tl::expected replace_frames(const std::string & fault_code, + const std::vector & frames) = 0; + + /// Drop every row for @p fault_code. Removing a code that has no rows is + /// not an error: the caller evicts by code and does not track what is on disk. + virtual tl::expected erase_frames(const std::string & fault_code) = 0; + + /// Every row, oldest capture first. The order is what lets a caller honour a + /// retained-frame bound by keeping the newest codes. + virtual tl::expected, std::string> load_all() = 0; +}; + +/// In-memory backend: the store contract without a file, for tests and for +/// callers that want the interface without persistence. +class InMemoryEntityFreezeFrameStore : public EntityFreezeFrameStore { + public: + tl::expected replace_frames(const std::string & fault_code, + const std::vector & frames) override { + std::lock_guard lock(mutex_); + if (frames.empty()) { + rows_.erase(fault_code); + return {}; + } + rows_[fault_code] = frames; + return {}; + } + + tl::expected erase_frames(const std::string & fault_code) override { + std::lock_guard lock(mutex_); + rows_.erase(fault_code); + return {}; + } + + tl::expected, std::string> load_all() override { + std::lock_guard lock(mutex_); + std::vector all; + for (const auto & entry : rows_) { + all.insert(all.end(), entry.second.begin(), entry.second.end()); + } + // Same total order the SQLite backend serves, so a caller's bound-keeping + // behaves identically on both. + std::stable_sort(all.begin(), all.end(), [](const StoredEntityFreezeFrame & a, const StoredEntityFreezeFrame & b) { + if (a.captured_at_ns != b.captured_at_ns) { + return a.captured_at_ns < b.captured_at_ns; + } + if (a.fault_code != b.fault_code) { + return a.fault_code < b.fault_code; + } + return a.entity_id < b.entity_id; + }); + return all; + } + + private: + mutable std::mutex mutex_; + std::map> rows_; +}; + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/sqlite_entity_freeze_frame_store.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/sqlite_entity_freeze_frame_store.hpp new file mode 100644 index 000000000..5b076b62a --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/sqlite_entity_freeze_frame_store.hpp @@ -0,0 +1,62 @@ +// 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. + +#pragma once + +#include + +#include +#include +#include + +#include "ros2_medkit_gateway/core/entity_freeze_frame_store.hpp" + +namespace ros2_medkit_gateway { + +/// SQLite-backed entity freeze-frame persistence. +/// +/// Thread-safe via internal mutex. The table is created on first open. Use +/// ":memory:" for an ephemeral database. +class SqliteEntityFreezeFrameStore : public EntityFreezeFrameStore { + public: + /// Open (or create) the database at `db_path`. + /// @throws std::runtime_error on SQLite open/init failure. + explicit SqliteEntityFreezeFrameStore(const std::string & db_path); + + ~SqliteEntityFreezeFrameStore() override; + + // Non-copyable, non-movable (owns SQLite connection) + SqliteEntityFreezeFrameStore(const SqliteEntityFreezeFrameStore &) = delete; + SqliteEntityFreezeFrameStore & operator=(const SqliteEntityFreezeFrameStore &) = delete; + SqliteEntityFreezeFrameStore(SqliteEntityFreezeFrameStore &&) = delete; + SqliteEntityFreezeFrameStore & operator=(SqliteEntityFreezeFrameStore &&) = delete; + + tl::expected replace_frames(const std::string & fault_code, + const std::vector & frames) override; + tl::expected erase_frames(const std::string & fault_code) override; + tl::expected, std::string> load_all() override; + + private: + /// Create the table if it does not exist. + void initialize_schema(); + + /// Delete every row for a code. Caller holds mutex_. + tl::expected delete_code_locked(const std::string & fault_code); + + std::string db_path_; + sqlite3 * db_{nullptr}; + mutable std::mutex mutex_; +}; + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/entity_freeze_frame_capture.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/entity_freeze_frame_capture.hpp index d3c041ee4..f86369767 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/entity_freeze_frame_capture.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/entity_freeze_frame_capture.hpp @@ -31,6 +31,7 @@ #include #include "rclcpp/rclcpp.hpp" +#include "ros2_medkit_gateway/core/entity_freeze_frame_store.hpp" #include "ros2_medkit_gateway/core/providers/data_provider.hpp" #include "ros2_medkit_gateway/ros2_common/ros2_subscription_slot.hpp" #include "ros2_medkit_msgs/msg/fault_event.hpp" @@ -57,6 +58,13 @@ namespace ros2_medkit_gateway { * kept across EVENT_CLEARED (the confirmed-state record stays attached to * the cleared fault's detail) and overwritten on every EVENT_CONFIRMED, so * a re-occurrence re-samples the plugin at its own confirm time. + * + * With an EntityFreezeFrameStore the frames also survive the process. Without + * one a restart loses them, and the startup catch-up then re-reads the plant + * as it is *now* and stamps that re-read with the restart - the values at + * fault time are gone. With one, the stored frame is reloaded before the + * catch-up and served exactly as it was captured, and the catch-up re-reads + * only the faults that have no frame. */ class EntityFreezeFrameCapture { public: @@ -67,6 +75,13 @@ class EntityFreezeFrameCapture { static constexpr const char * kSourceDataProvider = "plugin_data_provider"; static constexpr const char * kSourceXPlcDataRoute = "plugin_x_plc_data_route"; + /// Persisted marker for a startup catch-up frame, served as + /// ``x-medkit.capture_origin``. Stored so a reloaded frame keeps saying + /// which clock its captured_at came from: the property is intrinsic to the + /// frame, so a restart must not launder a catch-up frame into a confirm-edge + /// one. A confirm-edge frame stores the empty string and stays unmarked. + static constexpr const char * kCaptureOriginStartup = "startup"; + /// One captured frame: the entity's data values at fault-confirm time. /// captured_at_ns dates the capture, not the values - a disconnected entity /// serves its last known values, whose age is bounded only by the outage. @@ -104,6 +119,15 @@ class EntityFreezeFrameCapture { struct StandingFault { std::string fault_code; std::vector reporting_sources; + /// When THIS occurrence of the fault started, from the list reply's + /// `first_occurred` (seconds on the wire, nanoseconds here). The + /// fault_manager resets it only when a CLEARED fault reactivates, so it is + /// what tells a stored frame from a previous occurrence apart from one + /// belonging to the occurrence being served now. `last_occurred` cannot do + /// this: it moves on every report, so a fault that keeps failing would look + /// re-occurred at every restart. 0 when the reply did not report it, which + /// reads as "cannot tell" and keeps the stored frame. + int64_t first_occurred_ns{0}; }; /// Lists the faults that are already confirmed when this object starts, so @@ -114,6 +138,15 @@ class EntityFreezeFrameCapture { /// that is what lets the destructor's join interrupt the wait. using StandingFaultLister = std::function(const std::function & should_abort)>; + /// Reports every fault code the fault_manager still holds, whatever its + /// status, so a reloaded frame for a fault that is gone can be dropped. + /// Returns nullopt when the fault_manager could not be asked - the caller + /// then drops nothing, because "could not tell" must never read as "gone". + /// Called once, on the capture thread, right after the standing-fault + /// lister has already waited for the services. + using KnownFaultCodeLister = + std::function>(const std::function & should_abort)>; + /** * @param node ROS 2 node used to resolve the fault-events topic name and logger * @param exec shared subscription executor; the fault-events subscription is @@ -123,11 +156,19 @@ class EntityFreezeFrameCapture { * @param resolver entity-to-DataProvider resolver (typically wraps PluginManager) * @param route_fetcher x-plc-data route fallback for entities whose plugin * has no DataProvider (the commercial PLC bridges); may be null - * @param max_faults retained-frame bound; oldest fault's frames evicted past it + * @param max_faults retained-frame bound; oldest fault's frames evicted past it. + * Counts reloaded and freshly captured frames together. + * @param standing_lister lists the faults already confirmed at startup + * @param store frame persistence; null keeps the frames in process memory only + * @param known_code_lister reports the fault codes the fault_manager still + * holds, so reloaded frames for faults that are gone can be dropped; + * null keeps every reloaded frame */ EntityFreezeFrameCapture(rclcpp::Node * node, ros2_common::Ros2SubscriptionExecutor & exec, DataProviderResolver resolver, RouteDataFetcher route_fetcher = nullptr, - size_t max_faults = 256, StandingFaultLister standing_lister = nullptr); + size_t max_faults = 256, StandingFaultLister standing_lister = nullptr, + std::shared_ptr store = nullptr, + KnownFaultCodeLister known_code_lister = nullptr); ~EntityFreezeFrameCapture(); @@ -214,6 +255,56 @@ class EntityFreezeFrameCapture { /// clear/re-report cycle; one line per code is enough for an operator). void log_fallback_failure_once(const std::string & fault_code, const std::string & message); + /// Fill frames_ from the store. Runs in the constructor, so a frame captured + /// before the last shutdown is already being served when the first request + /// arrives, and the catch-up (which starts later, on capture_thread_) + /// already sees it. Honours max_faults_ by keeping the newest codes: a store + /// larger than the bound must not evict what it just loaded. + void load_persisted_frames(); + + /// Serialize a frame for the store. The columns carry entity, timestamp, + /// source and origin; the blob carries the values and the payload + /// provenance, so a reload reproduces the frame exactly. + static StoredEntityFreezeFrame to_stored(const std::string & fault_code, const Frame & frame); + + /// Inverse of to_stored. Returns nullopt for a row whose blob is not shaped + /// like a frame (a hand-edited or half-written file). + static std::optional from_stored(const StoredEntityFreezeFrame & row); + + /// Write the code's frames through to the store, replacing what was there. + /// Caller holds mutex_, so the file and the map cannot disagree. + void persist_frames_locked(const std::string & fault_code, const std::vector & frames); + + /// Drop the code's rows from the store. Caller holds mutex_. + void erase_persisted_locked(const std::string & fault_code); + + /// Drop reloaded frames whose fault the fault_manager no longer holds (its + /// store was replaced or wiped under ours). Codes reported in any status, + /// cleared included, are kept: a cleared fault keeps its frame. Does nothing + /// without a known-code lister, or when the lister cannot answer. + void prune_frames_for_unknown_faults(const std::function & should_abort); + + /// Reloaded codes whose stored frame belongs to an EARLIER occurrence of + /// their fault: it cleared and confirmed again while the gateway was down, so + /// the frame holds the previous incident's values and serving it unmarked + /// would present them as this occurrence's. + /// + /// Read-only on purpose. The row stays until a re-read has actually been + /// tried, so a fault whose entity answers gets its frame replaced rather than + /// deleted and then not re-taken. It is dropped only by drop_stale_frame(), + /// after the attempt failed. Only reloaded codes are eligible: a frame this + /// process captured is by definition this occurrence's. + std::unordered_set stale_reloaded_codes(const std::vector & standing) const; + + /// Last resort for a stale-occurrence code the catch-up could not re-read: + /// erase it from memory and from the store, and say so. Keeping it would + /// serve the previous incident's values unmarked, which is the defect this + /// path exists to fix, so the frame goes. Never silently, though: the + /// operator is losing evidence and only the log can tell them. + /// @p entities names the reporting sources that were tried (empty when the + /// fault named none), @p reason why no frame could be taken. + void drop_stale_frame(const std::string & fault_code, const std::string & entities, const char * reason); + std::unique_ptr subscription_slot_; DataProviderResolver resolver_; RouteDataFetcher route_fetcher_; @@ -230,6 +321,14 @@ class EntityFreezeFrameCapture { std::unordered_map> frames_; std::deque insertion_order_; ///< eviction order (FIFO) std::unordered_set fallback_logged_; ///< fault codes already warned about (bounded) + /// Codes whose frames came from the store and have not been re-captured + /// since. Only these are eligible for the unknown-fault prune: a frame this + /// process captured is by definition for a fault the fault_manager just + /// confirmed, whatever a stale list reply says. + std::unordered_set reloaded_codes_; + /// Store-write failures already warned about, so a read-only or full volume + /// costs one line, not one per capture. + bool store_write_warned_{false}; /// Confirm events pending capture: fed by the subscription worker, drained /// by capture_thread_. Bounded - oldest event dropped when full. @@ -241,6 +340,12 @@ class EntityFreezeFrameCapture { /// Lists faults already confirmed at construction; run once on that thread. StandingFaultLister standing_lister_; + + /// Frame persistence; null keeps the frames in process memory only. + std::shared_ptr store_; + /// Reports the codes the fault_manager still holds; run once, on the + /// capture thread, after the standing-fault lister. + KnownFaultCodeLister known_code_lister_; }; } // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp index 89a153d1b..25813eee0 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp @@ -328,6 +328,12 @@ class GatewayNode : public rclcpp::Node { void refresh_cache(); void start_rest_server(); + /// Open the entity freeze-frame store from `entity_freeze_frame.storage.path`, + /// falling back to a file next to the trigger store. Returns nullptr when no + /// path resolves or the file cannot be opened - the frames are then process + /// memory only, which is what they were before the store existed. + std::shared_ptr open_entity_freeze_frame_store(); + /// Log a one-time discovery summary shortly after startup: discovered node / /// topic / entity counts, the REST URL and a sample curl. When no application /// nodes are visible, also warn loudly with the active ROS environment diff --git a/src/ros2_medkit_gateway/src/core/sqlite_entity_freeze_frame_store.cpp b/src/ros2_medkit_gateway/src/core/sqlite_entity_freeze_frame_store.cpp new file mode 100644 index 000000000..0eaccfc6a --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/sqlite_entity_freeze_frame_store.cpp @@ -0,0 +1,259 @@ +// 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. + +#include "ros2_medkit_gateway/core/sqlite_entity_freeze_frame_store.hpp" + +#include +#include +#include + +namespace ros2_medkit_gateway { + +namespace { + +/// RAII wrapper for SQLite prepared statements (mirrors SqliteTriggerStore's, +/// plus the 64-bit binds a nanosecond timestamp needs). +class SqliteStatement { + public: + SqliteStatement(sqlite3 * db, const char * sql) : db_(db) { + if (sqlite3_prepare_v2(db, sql, -1, &stmt_, nullptr) != SQLITE_OK) { + throw std::runtime_error(std::string("Failed to prepare statement: ") + sqlite3_errmsg(db)); + } + } + + ~SqliteStatement() { + if (stmt_) { + sqlite3_finalize(stmt_); + } + } + + SqliteStatement(const SqliteStatement &) = delete; + SqliteStatement & operator=(const SqliteStatement &) = delete; + SqliteStatement(SqliteStatement &&) = delete; + SqliteStatement & operator=(SqliteStatement &&) = delete; + + void bind_text(int index, const std::string & value) { + const auto size = value.size(); + if (size > static_cast(std::numeric_limits::max())) { + throw std::runtime_error("bind_text: value exceeds SQLite int length limit"); + } + if (sqlite3_bind_text(stmt_, index, value.c_str(), static_cast(size), SQLITE_TRANSIENT) != SQLITE_OK) { + throw std::runtime_error(std::string("Failed to bind text: ") + sqlite3_errmsg(db_)); + } + } + + void bind_int64(int index, int64_t value) { + if (sqlite3_bind_int64(stmt_, index, value) != SQLITE_OK) { + throw std::runtime_error(std::string("Failed to bind int64: ") + sqlite3_errmsg(db_)); + } + } + + int step() { + return sqlite3_step(stmt_); + } + + std::string column_text(int index) { + const auto * text = reinterpret_cast(sqlite3_column_text(stmt_, index)); + return text ? std::string(text) : std::string(); + } + + int64_t column_int64(int index) { + return sqlite3_column_int64(stmt_, index); + } + + private: + sqlite3 * db_; + sqlite3_stmt * stmt_{nullptr}; +}; + +} // namespace + +// --------------------------------------------------------------------------- +// Lifecycle +// --------------------------------------------------------------------------- + +SqliteEntityFreezeFrameStore::SqliteEntityFreezeFrameStore(const std::string & db_path) : db_path_(db_path) { + int flags = SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX; + if (sqlite3_open_v2(db_path.c_str(), &db_, flags, nullptr) != SQLITE_OK) { + std::string error = db_ ? sqlite3_errmsg(db_) : "Unknown error"; + if (db_) { + sqlite3_close(db_); + db_ = nullptr; + } + throw std::runtime_error("Failed to open entity freeze-frame database '" + db_path + "': " + error); + } + + char * err_msg = nullptr; + if (sqlite3_exec(db_, "PRAGMA journal_mode=WAL;", nullptr, nullptr, &err_msg) != SQLITE_OK) { + std::string error = err_msg ? err_msg : "Unknown error"; + sqlite3_free(err_msg); + sqlite3_close(db_); + db_ = nullptr; + throw std::runtime_error("Failed to enable WAL mode: " + error); + } + + sqlite3_busy_timeout(db_, 5000); + initialize_schema(); +} + +SqliteEntityFreezeFrameStore::~SqliteEntityFreezeFrameStore() { + if (db_) { + sqlite3_close(db_); + } +} + +// --------------------------------------------------------------------------- +// Schema +// --------------------------------------------------------------------------- + +void SqliteEntityFreezeFrameStore::initialize_schema() { + // One row per (fault_code, entity_id): a fault reported by two entities + // freezes both, and a re-confirm replaces the code's rows as a unit. + const char * create_table = R"( + CREATE TABLE IF NOT EXISTS entity_freeze_frames ( + fault_code TEXT NOT NULL, + entity_id TEXT NOT NULL, + frame TEXT NOT NULL, + captured_at_ns INTEGER NOT NULL, + source TEXT NOT NULL DEFAULT '', + capture_origin TEXT NOT NULL DEFAULT '', + PRIMARY KEY (fault_code, entity_id) + ); + )"; + + char * err_msg = nullptr; + if (sqlite3_exec(db_, create_table, nullptr, nullptr, &err_msg) != SQLITE_OK) { + std::string error = err_msg ? err_msg : "Unknown error"; + sqlite3_free(err_msg); + throw std::runtime_error("Failed to create entity_freeze_frames table: " + error); + } +} + +// --------------------------------------------------------------------------- +// Writes +// --------------------------------------------------------------------------- + +tl::expected SqliteEntityFreezeFrameStore::delete_code_locked(const std::string & fault_code) { + SqliteStatement del(db_, "DELETE FROM entity_freeze_frames WHERE fault_code = ?"); + del.bind_text(1, fault_code); + if (del.step() != SQLITE_DONE) { + return tl::make_unexpected(std::string("Failed to delete entity freeze-frames: ") + sqlite3_errmsg(db_)); + } + return {}; +} + +tl::expected +SqliteEntityFreezeFrameStore::replace_frames(const std::string & fault_code, + const std::vector & frames) { + std::lock_guard lock(mutex_); + + try { + // Delete-then-insert in one transaction: a re-confirm that no longer + // reports an entity must not leave that entity's stale row behind, and a + // reader must never see the code half-written. + char * err_msg = nullptr; + if (sqlite3_exec(db_, "BEGIN IMMEDIATE", nullptr, nullptr, &err_msg) != SQLITE_OK) { + std::string error = err_msg ? err_msg : "Unknown error"; + sqlite3_free(err_msg); + return tl::make_unexpected("replace_frames: BEGIN failed: " + error); + } + + const auto rollback = [this] { + sqlite3_exec(db_, "ROLLBACK", nullptr, nullptr, nullptr); + }; + + auto deleted = delete_code_locked(fault_code); + if (!deleted) { + rollback(); + return deleted; + } + + for (const auto & frame : frames) { + SqliteStatement stmt(db_, + "INSERT OR REPLACE INTO entity_freeze_frames " + "(fault_code, entity_id, frame, captured_at_ns, source, capture_origin) " + "VALUES (?,?,?,?,?,?)"); + stmt.bind_text(1, fault_code); + stmt.bind_text(2, frame.entity_id); + stmt.bind_text(3, frame.frame.dump()); + stmt.bind_int64(4, frame.captured_at_ns); + stmt.bind_text(5, frame.source); + stmt.bind_text(6, frame.capture_origin); + if (stmt.step() != SQLITE_DONE) { + std::string error = sqlite3_errmsg(db_); + rollback(); + return tl::make_unexpected("Failed to save entity freeze-frame: " + error); + } + } + + if (sqlite3_exec(db_, "COMMIT", nullptr, nullptr, &err_msg) != SQLITE_OK) { + std::string error = err_msg ? err_msg : "Unknown error"; + sqlite3_free(err_msg); + rollback(); + return tl::make_unexpected("replace_frames: COMMIT failed: " + error); + } + return {}; + } catch (const std::exception & e) { + sqlite3_exec(db_, "ROLLBACK", nullptr, nullptr, nullptr); + return tl::make_unexpected(std::string("replace_frames: ") + e.what()); + } +} + +tl::expected SqliteEntityFreezeFrameStore::erase_frames(const std::string & fault_code) { + std::lock_guard lock(mutex_); + try { + return delete_code_locked(fault_code); + } catch (const std::exception & e) { + return tl::make_unexpected(std::string("erase_frames: ") + e.what()); + } +} + +// --------------------------------------------------------------------------- +// load_all +// --------------------------------------------------------------------------- + +tl::expected, std::string> SqliteEntityFreezeFrameStore::load_all() { + std::lock_guard lock(mutex_); + + try { + SqliteStatement stmt(db_, + "SELECT fault_code, entity_id, frame, captured_at_ns, source, capture_origin " + "FROM entity_freeze_frames " + "ORDER BY captured_at_ns ASC, fault_code ASC, entity_id ASC"); + + std::vector result; + while (stmt.step() == SQLITE_ROW) { + StoredEntityFreezeFrame row; + row.fault_code = stmt.column_text(0); + row.entity_id = stmt.column_text(1); + auto parsed = nlohmann::json::parse(stmt.column_text(2), nullptr, false); + if (parsed.is_discarded()) { + // One unreadable row must not cost the operator every other frame: + // skip it, the caller's catch-up re-reads that fault if it is still + // standing. + continue; + } + row.frame = std::move(parsed); + row.captured_at_ns = stmt.column_int64(3); + row.source = stmt.column_text(4); + row.capture_origin = stmt.column_text(5); + result.push_back(std::move(row)); + } + return result; + } catch (const std::exception & e) { + return tl::make_unexpected(std::string("load_all: ") + e.what()); + } +} + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/entity_freeze_frame_capture.cpp b/src/ros2_medkit_gateway/src/entity_freeze_frame_capture.cpp index 01bde9674..d7de8333d 100644 --- a/src/ros2_medkit_gateway/src/entity_freeze_frame_capture.cpp +++ b/src/ros2_medkit_gateway/src/entity_freeze_frame_capture.cpp @@ -14,8 +14,10 @@ #include "ros2_medkit_gateway/entity_freeze_frame_capture.hpp" +#include #include #include +#include #include "ros2_medkit_gateway/fault_manager_paths.hpp" @@ -34,6 +36,10 @@ constexpr size_t kMaxLoggedFaultCodes = 1024; /// the standing-fault snapshot; past it the catch-up proceeds best-effort. constexpr std::chrono::seconds kEventsMatchTimeout{10}; +/// How many over-the-bound fault codes the truncation warning names before it +/// stops. Enough for an operator to act on, short of a 256-code log line. +constexpr size_t kMaxNamedOverCapCodes = 10; + /// Read a string field totally: json::value() throws type_error.302 when the /// key is present but not a string, and plugin content is untrusted. std::string string_field(const nlohmann::json & item, const char * field) { @@ -41,16 +47,37 @@ std::string string_field(const nlohmann::json & item, const char * field) { return it != item.end() && it->is_string() ? it->get() : std::string(); } +/// Reporting sources as one comma-separated string, for a log line that has to +/// name the entities an operator would go and look at. +std::string join_sources(const std::vector & sources) { + std::string joined; + for (const auto & source : sources) { + if (!joined.empty()) { + joined += ", "; + } + joined += source; + } + return joined; +} + } // namespace EntityFreezeFrameCapture::EntityFreezeFrameCapture(rclcpp::Node * node, ros2_common::Ros2SubscriptionExecutor & exec, DataProviderResolver resolver, RouteDataFetcher route_fetcher, - size_t max_faults, StandingFaultLister standing_lister) + size_t max_faults, StandingFaultLister standing_lister, + std::shared_ptr store, + KnownFaultCodeLister known_code_lister) : resolver_(std::move(resolver)) , route_fetcher_(std::move(route_fetcher)) , logger_(node->get_logger()) , max_faults_(max_faults > 0 ? max_faults : 1) - , standing_lister_(std::move(standing_lister)) { + , standing_lister_(std::move(standing_lister)) + , store_(std::move(store)) + , known_code_lister_(std::move(known_code_lister)) { + // Before anything can serve or capture: a frame taken before the last + // shutdown is the one the operator is owed, and the catch-up must see it so + // it re-reads only the faults that have none. + load_persisted_frames(); // Resolve the topic from the gateway node (it owns fault_manager.namespace); // the subscription itself is created on the executor's dedicated _sub node so // it never races rcl's hash-map on the main node (issue #375). @@ -104,6 +131,252 @@ EntityFreezeFrameCapture::frames_for(const std::string & fault_code) const { return it != frames_.end() ? it->second : std::vector{}; } +StoredEntityFreezeFrame EntityFreezeFrameCapture::to_stored(const std::string & fault_code, const Frame & frame) { + StoredEntityFreezeFrame row; + row.fault_code = fault_code; + row.entity_id = frame.entity_id; + row.frame = nlohmann::json::object(); + row.frame["values"] = frame.values; + // Written only when the capture had them, so the reload reproduces the + // frame's own "reported nothing" as absence rather than as a null. + if (frame.connected.has_value()) { + row.frame["connected"] = *frame.connected; + } + if (!frame.source_timestamp.is_null()) { + row.frame["source_timestamp"] = frame.source_timestamp; + } + row.captured_at_ns = frame.captured_at_ns; + row.source = frame.source; + row.capture_origin = frame.startup_catchup ? kCaptureOriginStartup : ""; + return row; +} + +std::optional +EntityFreezeFrameCapture::from_stored(const StoredEntityFreezeFrame & row) { + if (row.entity_id.empty() || !row.frame.is_object()) { + return std::nullopt; + } + const auto values = row.frame.find("values"); + if (values == row.frame.end()) { + return std::nullopt; + } + Frame frame; + frame.entity_id = row.entity_id; + frame.values = *values; + frame.captured_at_ns = row.captured_at_ns; + frame.source = row.source; + // Reloading must not launder a catch-up frame into a confirm-edge one: its + // captured_at is still a gateway start, so the marker still belongs on it. + frame.startup_catchup = row.capture_origin == kCaptureOriginStartup; + const auto connected = row.frame.find("connected"); + if (connected != row.frame.end() && connected->is_boolean()) { + frame.connected = connected->get(); + } + const auto source_timestamp = row.frame.find("source_timestamp"); + if (source_timestamp != row.frame.end()) { + frame.source_timestamp = *source_timestamp; + } + return frame; +} + +void EntityFreezeFrameCapture::persist_frames_locked(const std::string & fault_code, + const std::vector & frames) { + if (!store_) { + return; + } + std::vector rows; + rows.reserve(frames.size()); + for (const auto & frame : frames) { + rows.push_back(to_stored(fault_code, frame)); + } + auto written = store_->replace_frames(fault_code, rows); + if (!written && !store_write_warned_) { + // One line for the life of the process: a read-only or full volume would + // otherwise log once per confirm, and the frames still work in memory. + store_write_warned_ = true; + RCLCPP_WARN(logger_, "Entity freeze-frame store write failed, frames are process-local until restart: %s", + written.error().c_str()); + } +} + +void EntityFreezeFrameCapture::erase_persisted_locked(const std::string & fault_code) { + if (!store_) { + return; + } + auto erased = store_->erase_frames(fault_code); + if (!erased && !store_write_warned_) { + store_write_warned_ = true; + RCLCPP_WARN(logger_, "Entity freeze-frame store delete failed: %s", erased.error().c_str()); + } +} + +void EntityFreezeFrameCapture::load_persisted_frames() { + if (!store_) { + return; + } + auto rows = store_->load_all(); + if (!rows) { + RCLCPP_WARN(logger_, "Entity freeze-frame store unreadable, starting with no reloaded frames: %s", + rows.error().c_str()); + return; + } + + std::unordered_map> loaded; + std::unordered_map newest; + size_t unreadable = 0; + for (const auto & row : *rows) { + auto frame = from_stored(row); + if (!frame) { + ++unreadable; + continue; + } + auto it = newest.find(row.fault_code); + if (it == newest.end()) { + newest.emplace(row.fault_code, row.captured_at_ns); + } else { + it->second = std::max(it->second, row.captured_at_ns); + } + loaded[row.fault_code].push_back(std::move(*frame)); + } + + // Oldest code first, so the retained-frame bound drops what a restart can + // least afford to keep rather than what it just read. + std::vector codes; + codes.reserve(loaded.size()); + for (const auto & entry : loaded) { + codes.push_back(entry.first); + } + std::sort(codes.begin(), codes.end(), [&newest](const std::string & a, const std::string & b) { + if (newest.at(a) != newest.at(b)) { + return newest.at(a) < newest.at(b); + } + return a < b; + }); + const size_t over_cap = codes.size() > max_faults_ ? codes.size() - max_faults_ : 0; + + std::lock_guard lock(mutex_); + for (size_t i = 0; i < codes.size(); ++i) { + const auto & code = codes[i]; + if (i < over_cap) { + // Past the bound: drop the rows too, or every start re-reads frames it + // can never serve and the file grows without one. + erase_persisted_locked(code); + continue; + } + insertion_order_.push_back(code); + frames_[code] = std::move(loaded[code]); + reloaded_codes_.insert(code); + } + if (!frames_.empty()) { + RCLCPP_INFO(logger_, "Entity freeze-frame: reloaded frames for %zu fault(s) from the store", frames_.size()); + } + if (over_cap > 0) { + RCLCPP_WARN(logger_, + "Entity freeze-frame store held %zu fault(s) beyond the retained-frame bound of %zu; " + "the oldest were dropped", + over_cap, max_faults_); + } + if (unreadable > 0) { + RCLCPP_WARN(logger_, "Entity freeze-frame store: %zu unreadable row(s) skipped", unreadable); + } +} + +void EntityFreezeFrameCapture::prune_frames_for_unknown_faults(const std::function & should_abort) { + if (!known_code_lister_) { + return; + } + { + std::lock_guard lock(mutex_); + if (reloaded_codes_.empty()) { + return; + } + } + std::optional> known; + try { + known = known_code_lister_(should_abort); + } catch (const std::exception & e) { + RCLCPP_WARN(logger_, "Entity freeze-frame prune skipped: known-fault lister threw: %s", e.what()); + return; + } catch (...) { + RCLCPP_WARN(logger_, "Entity freeze-frame prune skipped: known-fault lister threw"); + return; + } + if (!known) { + return; // could not ask: "cannot tell" must never read as "the fault is gone" + } + + size_t dropped = 0; + { + std::lock_guard lock(mutex_); + for (auto it = reloaded_codes_.begin(); it != reloaded_codes_.end();) { + if (known->count(*it) != 0) { + ++it; // reported in any status, cleared included: the frame stays + continue; + } + const std::string code = *it; + it = reloaded_codes_.erase(it); + frames_.erase(code); + insertion_order_.erase(std::remove(insertion_order_.begin(), insertion_order_.end(), code), + insertion_order_.end()); + erase_persisted_locked(code); + ++dropped; + } + } + if (dropped > 0) { + RCLCPP_INFO(logger_, + "Entity freeze-frame: dropped %zu reloaded frame(s) for fault(s) the fault manager no longer holds", + dropped); + } +} + +std::unordered_set +EntityFreezeFrameCapture::stale_reloaded_codes(const std::vector & standing) const { + std::unordered_set stale; + std::lock_guard lock(mutex_); + if (reloaded_codes_.empty()) { + return stale; + } + for (const auto & fault : standing) { + if (fault.first_occurred_ns <= 0 || reloaded_codes_.count(fault.fault_code) == 0) { + continue; + } + const auto entry = frames_.find(fault.fault_code); + if (entry == frames_.end()) { + continue; + } + // The newest of the code's frames dates the stored capture: they are all + // written by one capture, so if even that one predates the occurrence the + // whole set belongs to an incident that has since been cleared. + int64_t newest = 0; + for (const auto & frame : entry->second) { + newest = std::max(newest, frame.captured_at_ns); + } + if (fault.first_occurred_ns <= newest) { + continue; // same occurrence: the stored frame is the one to serve + } + stale.insert(fault.fault_code); + } + return stale; +} + +void EntityFreezeFrameCapture::drop_stale_frame(const std::string & fault_code, const std::string & entities, + const char * reason) { + { + std::lock_guard lock(mutex_); + frames_.erase(fault_code); + insertion_order_.erase(std::remove(insertion_order_.begin(), insertion_order_.end(), fault_code), + insertion_order_.end()); + reloaded_codes_.erase(fault_code); + erase_persisted_locked(fault_code); + } + // The operator is losing evidence here. Keeping the frame would serve the + // previous incident's values as this one's, so it goes, but never silently. + RCLCPP_WARN(logger_, + "Entity freeze-frame for fault '%s': the stored frame is from an earlier occurrence and entity '%s' " + "could not be re-read (%s). The stored frame was discarded, so this occurrence has no freeze-frame.", + fault_code.c_str(), entities.c_str(), reason); +} + nlohmann::json EntityFreezeFrameCapture::values_from_list_content(const nlohmann::json & content) { if (!content.contains("items") || !content["items"].is_array()) { return content; @@ -181,6 +454,23 @@ EntityFreezeFrameCapture::standing_faults_from_list_reply(const nlohmann::json & fault.reporting_sources.push_back(src.get()); } } + // Seconds on the wire (fault_msg_conversions), nanoseconds here so it can + // be compared with a frame's captured_at_ns without converting per row. + // Anything that is not a positive number leaves it at 0, which reads as + // "the reply did not say" and never costs a stored frame. + const auto first_occurred = item.find("first_occurred"); + if (first_occurred != item.end() && first_occurred->is_number()) { + const double seconds = first_occurred->get(); + // The range is checked on the nanosecond product, before the cast: a + // double outside int64's range makes the conversion undefined, and a + // NaN fails every comparison so it lands here too. Untrusted input on + // this path is a malformed or hostile reply, not just a stale clock. + const double nanoseconds = seconds * 1e9; + constexpr double kMaxRepresentableNs = 9.2e18; // below int64 max, with room for the ulp + if (nanoseconds > 0.0 && nanoseconds < kMaxRepresentableNs) { + fault.first_occurred_ns = static_cast(nanoseconds); + } + } standing.push_back(std::move(fault)); } return standing; @@ -288,7 +578,7 @@ void EntityFreezeFrameCapture::wait_for_events_publisher(const std::function standing; - try { - standing = standing_lister_(should_abort); - } catch (const std::exception & e) { - RCLCPP_WARN(logger_, "Entity freeze-frame startup catch-up failed: standing-fault lister threw: %s", e.what()); - return; - } catch (...) { - RCLCPP_WARN(logger_, "Entity freeze-frame startup catch-up failed: standing-fault lister threw"); + if (standing_lister_) { + try { + standing = standing_lister_(should_abort); + } catch (const std::exception & e) { + RCLCPP_WARN(logger_, "Entity freeze-frame startup catch-up failed: standing-fault lister threw: %s", e.what()); + return; + } catch (...) { + RCLCPP_WARN(logger_, "Entity freeze-frame startup catch-up failed: standing-fault lister threw"); + return; + } + } + // Runs after the lister has already waited the fault services out, so the + // extra query costs a round trip rather than a second startup stall. + prune_frames_for_unknown_faults(should_abort); + if (!standing_lister_ || should_abort()) { return; } // Codes with a confirm already queued belong to the drain loop: capturing @@ -327,8 +625,85 @@ void EntityFreezeFrameCapture::capture_standing_faults() { queued_codes.insert(queued->fault.fault_code); } } - size_t framed = 0; + // Reloaded frames from an occurrence that has since been cleared and + // re-confirmed. Identified here, still on disk: whether the row goes is + // decided below, by whether the re-read could replace it. + const auto stale = stale_reloaded_codes(standing); + + size_t captured = 0; + size_t re_read = 0; + size_t discarded = 0; size_t over_cap = 0; + + // ---- Pass 1: the stale codes, in place ---------------------------------- + // Each already owns a slot in frames_, and capture_for_event evicts only when + // the code is new to the map, so a re-read can neither exceed the + // retained-frame bound nor push anyone else out. That makes the bound check + // wrong here in both directions. It would refuse a read that costs nothing, + // and while these codes are counted as absent a bare code admitted against + // that under-count would FIFO-evict a live frame that is still wanted. So the + // stale codes are settled first, and only then is the real occupancy known. + // + // Every code this pass touches is recorded, replaced or dropped alike. A + // replaced one is back in frames_ and pass 2 would skip it anyway, but a + // dropped one is not, and without this pass 2 would see a fault with no frame + // and a free slot and ask the same unreachable entity a second time. One + // blocking plugin read per catch-up per fault, and no more. + std::unordered_set settled; + for (const auto & fault : standing) { + if (should_abort()) { + return; + } + if (fault.fault_code.empty() || stale.count(fault.fault_code) == 0) { + continue; + } + // A stale code also jumps the queued-confirm skip. That skip exists so one + // confirm costs one plugin read, but here the alternative is leaving a + // frame from a dead occurrence in place on the chance the drain loop + // succeeds. One extra read is the cheaper mistake. + if (fault.reporting_sources.empty()) { + // The staleness test and the re-read test must agree on this, or a fault + // with no entity is stale to one and invisible to the other, and its row + // survives to be served. + drop_stale_frame(fault.fault_code, "", "the fault reports no entity to read"); + settled.insert(fault.fault_code); + ++discarded; + continue; + } + settled.insert(fault.fault_code); + ros2_medkit_msgs::msg::FaultEvent event; + event.event_type = ros2_medkit_msgs::msg::FaultEvent::EVENT_CONFIRMED; + event.fault.fault_code = fault.fault_code; + event.fault.reporting_sources = fault.reporting_sources; + // capture_for_event replaces the code's frames and its rows as one + // delete-then-insert, so a successful re-read swaps the stale frame out + // without a window in which the fault has none. + if (capture_for_event(event, /*startup_catchup=*/true)) { + ++captured; + ++re_read; + } else { + drop_stale_frame(fault.fault_code, join_sources(fault.reporting_sources), "the entity served no usable values"); + ++discarded; // the slot it held is now free for the NEXT fault, not for this one again + } + } + + // ---- Pass 2: the faults that have no frame at all ------------------------ + // Occupancy is read after the stale pass, so it is what frames_ really holds: + // every stale code has by now been replaced in place or dropped. A bare code + // is therefore admitted only against a slot that is genuinely free, and the + // FIFO inside capture_for_event can only reach codes that are neither stale + // nor mid-re-read. + std::vector over_cap_codes; + std::unordered_set already_framed; + size_t framed = 0; + { + std::lock_guard lock(mutex_); + already_framed.reserve(frames_.size()); + for (const auto & entry : frames_) { + already_framed.insert(entry.first); + } + framed = frames_.size(); + } for (const auto & fault : standing) { if (should_abort()) { return; @@ -336,11 +711,31 @@ void EntityFreezeFrameCapture::capture_standing_faults() { if (fault.fault_code.empty() || fault.reporting_sources.empty()) { continue; } + // Codes with a confirm already queued belong to the drain loop: capturing + // them here too would read the plugin twice for one confirm. if (queued_codes.count(fault.fault_code) != 0) { continue; } + // The stored frame is the one from this fault's own confirm edge. Re-reading + // the plant now would replace it with today's values under a "startup" + // marker, which is exactly what persisting the frame is here to stop. + // + // `settled` is the other half of that: pass 1 has already had its one go at + // every stale code, and a code it dropped is absent from frames_ precisely + // because its entity could not answer. Asking again in the same catch-up + // would be a second blocking read of an entity that just failed, and if it + // answered this time the fault would end up holding a frame the warning has + // already said it discarded. + if (already_framed.count(fault.fault_code) != 0 || settled.count(fault.fault_code) != 0) { + continue; + } if (framed >= max_faults_) { - ++over_cap; // storing more would FIFO-evict this catch-up's own frames + ++over_cap; // storing more would FIFO-evict a frame that is still wanted + // Named, not just counted: "3 faults went unframed" leaves an operator + // with no way to tell which fault details are missing their context. + if (over_cap_codes.size() < kMaxNamedOverCapCodes) { + over_cap_codes.push_back(fault.fault_code); + } continue; } ros2_medkit_msgs::msg::FaultEvent event; @@ -349,16 +744,24 @@ void EntityFreezeFrameCapture::capture_standing_faults() { event.fault.reporting_sources = fault.reporting_sources; if (capture_for_event(event, /*startup_catchup=*/true)) { ++framed; + ++captured; } } if (over_cap > 0) { RCLCPP_WARN(logger_, "Entity freeze-frame startup catch-up truncated: %zu standing fault(s) beyond the retained-frame " - "bound of %zu", - over_cap, max_faults_); + "bound of %zu, so they have no freeze-frame: %s%s", + over_cap, max_faults_, join_sources(over_cap_codes).c_str(), + over_cap > over_cap_codes.size() ? ", ..." : ""); + } + if (captured > 0) { + RCLCPP_INFO(logger_, "Entity freeze-frame: captured %zu fault(s) that were already confirmed at startup", captured); } - if (framed > 0) { - RCLCPP_INFO(logger_, "Entity freeze-frame: captured %zu fault(s) that were already confirmed at startup", framed); + if (re_read > 0 || discarded > 0) { + RCLCPP_INFO(logger_, + "Entity freeze-frame: %zu reloaded frame(s) from an earlier occurrence re-read, %zu discarded with no " + "replacement", + re_read, discarded); } } @@ -432,11 +835,23 @@ bool EntityFreezeFrameCapture::capture_for_event(const ros2_medkit_msgs::msg::Fa if (frames_.find(fault_code) == frames_.end()) { insertion_order_.push_back(fault_code); while (frames_.size() >= max_faults_ && !insertion_order_.empty()) { - frames_.erase(insertion_order_.front()); + const std::string evicted = insertion_order_.front(); + frames_.erase(evicted); insertion_order_.pop_front(); + // The store follows the map out: an evicted frame that stayed on disk + // would come back on the next start and the bound would mean nothing + // across restarts. + reloaded_codes_.erase(evicted); + erase_persisted_locked(evicted); } } + // A capture on this fault's own edge supersedes whatever was reloaded for it, + // so the code is no longer a candidate for the reloaded-frame prune. + reloaded_codes_.erase(fault_code); frames_[fault_code] = std::move(frames); + // Under the same lock as the map, so the file and what is being served + // cannot disagree about what was frozen. + persist_frames_locked(fault_code, frames_[fault_code]); RCLCPP_DEBUG(logger_, "Captured entity freeze-frame(s) for fault '%s'", fault_code.c_str()); return true; diff --git a/src/ros2_medkit_gateway/src/gateway_node.cpp b/src/ros2_medkit_gateway/src/gateway_node.cpp index 0d4e8c5ff..e89699fc1 100644 --- a/src/ros2_medkit_gateway/src/gateway_node.cpp +++ b/src/ros2_medkit_gateway/src/gateway_node.cpp @@ -21,7 +21,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -41,6 +43,7 @@ #include "ros2_medkit_gateway/plugins/ros_plugin_context.hpp" #include "ros2_medkit_gateway/core/http/handlers/sse_transport_provider.hpp" +#include "ros2_medkit_gateway/core/sqlite_entity_freeze_frame_store.hpp" #include "ros2_medkit_gateway/core/sqlite_trigger_store.hpp" using namespace std::chrono_literals; @@ -232,6 +235,7 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki // Zero-config freeze-frames for plugin-backed entities (opt-out) declare_parameter("entity_freeze_frame.enabled", true); + declare_parameter("entity_freeze_frame.storage.path", ""); // Locking parameters declare_parameter("locking.enabled", true); @@ -1871,6 +1875,7 @@ void GatewayNode::init_entity_freeze_frame_capture(ros2_common::Ros2Subscription if (!get_parameter("entity_freeze_frame.enabled").as_bool() || !plugin_mgr_ || !plugin_mgr_->has_plugins()) { return; } + auto frame_store = open_entity_freeze_frame_store(); entity_freeze_frame_capture_ = std::make_unique( this, exec, [this](const std::string & entity_id) { @@ -1927,9 +1932,71 @@ void GatewayNode::init_entity_freeze_frame_capture(ros2_common::Ros2Subscription RCLCPP_INFO(get_logger(), "Standing-fault freeze-frame catch-up: no confirmed faults at startup"); } return std::move(*parsed); + }, + std::move(frame_store), + // Every code the fault manager still holds, in any status. Frames for + // faults it no longer has (its own store was replaced under ours) are + // dropped; nullopt means it could not be asked, and then nothing is. + [this](const std::function & should_abort) -> std::optional> { + if (!fault_service_transport_ || should_abort()) { + return std::nullopt; + } + // Non-blocking: the standing-fault lister has already waited the + // services out, so a miss here means they are genuinely absent and + // the reply would say nothing about what still exists. + if (!fault_service_transport_->is_available()) { + return std::nullopt; + } + auto result = fault_service_transport_->list_faults("", true, true, true, true, true, false); + if (!result.success) { + return std::nullopt; + } + const auto faults = result.data.find("faults"); + if (faults == result.data.end() || !faults->is_array()) { + return std::nullopt; + } + std::unordered_set codes; + for (const auto & item : *faults) { + if (item.is_object()) { + const auto code = item.find("fault_code"); + if (code != item.end() && code->is_string()) { + codes.insert(code->get()); + } + } + } + return codes; }); } +std::shared_ptr GatewayNode::open_entity_freeze_frame_store() { + // An explicit path wins. With none, the frames go next to the trigger store, + // which is where an operator already points a persistent volume; with no + // trigger store either there is nowhere to put them and they stay in memory, + // as they were before the store existed. + std::string path = get_parameter("entity_freeze_frame.storage.path").as_string(); + if (path.empty()) { + const std::string trigger_path = get_parameter("triggers.storage.path").as_string(); + if (trigger_path.empty()) { + RCLCPP_INFO(get_logger(), + "Entity freeze-frames are not persisted (no entity_freeze_frame.storage.path and no " + "triggers.storage.path); they are lost on restart"); + return nullptr; + } + path = (std::filesystem::path(trigger_path).parent_path() / "entity_freeze_frames.db").string(); + } + try { + auto store = std::make_shared(path); + RCLCPP_INFO(get_logger(), "Entity freeze-frames persisted in %s", path.c_str()); + return store; + } catch (const std::exception & e) { + // A store the gateway cannot open must not stop it from capturing: the + // frames stay in memory, exactly as they did before persistence existed. + RCLCPP_ERROR(get_logger(), "Entity freeze-frame store '%s' could not be opened, frames stay in memory: %s", + path.c_str(), e.what()); + return nullptr; + } +} + EntityFreezeFrameCapture * GatewayNode::get_entity_freeze_frame_capture() const { return entity_freeze_frame_capture_.get(); } diff --git a/src/ros2_medkit_gateway/test/test_entity_freeze_frame_capture.cpp b/src/ros2_medkit_gateway/test/test_entity_freeze_frame_capture.cpp index b97372530..cbce202ef 100644 --- a/src/ros2_medkit_gateway/test/test_entity_freeze_frame_capture.cpp +++ b/src/ros2_medkit_gateway/test/test_entity_freeze_frame_capture.cpp @@ -14,19 +14,26 @@ #include +#include #include #include #include #include +#include +#include #include +#include #include #include #include +#include #include #include #include +#include #include +#include "ros2_medkit_gateway/core/entity_freeze_frame_store.hpp" #include "ros2_medkit_gateway/entity_freeze_frame_capture.hpp" #include "ros2_medkit_gateway/http/handlers/fault_handlers.hpp" #include "ros2_medkit_gateway/ros2_common/ros2_subscription_executor.hpp" @@ -38,6 +45,8 @@ using ros2_medkit_gateway::DataProvider; using ros2_medkit_gateway::DataProviderError; using ros2_medkit_gateway::DataProviderErrorInfo; using ros2_medkit_gateway::EntityFreezeFrameCapture; +using ros2_medkit_gateway::InMemoryEntityFreezeFrameStore; +using ros2_medkit_gateway::StoredEntityFreezeFrame; using ros2_medkit_gateway::handlers::FaultHandlers; using ros2_medkit_gateway::ros2_common::Ros2SubscriptionExecutor; using ros2_medkit_msgs::msg::Fault; @@ -676,6 +685,897 @@ TEST_F(EntityFreezeFrameCaptureTest, OldestFaultEvictedPastMaxFaults) { EXPECT_FALSE(capture.frames_for("PLC_EVICT_C").empty()); } +// =========================================================================== +// Persistence: the frame outlives the process, so a restart serves what was +// frozen at fault time instead of re-reading the plant as it is now. +// =========================================================================== + +namespace { + +/// One stored row, shaped as the capture writes them. +StoredEntityFreezeFrame make_stored_row(const std::string & fault_code, const std::string & entity_id, + int64_t captured_at_ns, double level = 7.0, + const std::string & capture_origin = "") { + StoredEntityFreezeFrame row; + row.fault_code = fault_code; + row.entity_id = entity_id; + row.frame = json{{"values", {{"level", level}}}, {"connected", false}, {"source_timestamp", "2026-09-08T17:51:40Z"}}; + row.captured_at_ns = captured_at_ns; + row.source = EntityFreezeFrameCapture::kSourceXPlcDataRoute; + row.capture_origin = capture_origin; + return row; +} + +/// Store that counts what the capture asks of it, so a test can pin the ORDER +/// of a replacement. A stale frame must be swapped out by one write, never +/// erased first and re-taken afterwards if the plant happens to answer. +class CountingEntityFreezeFrameStore : public ros2_medkit_gateway::EntityFreezeFrameStore { + public: + tl::expected replace_frames(const std::string & fault_code, + const std::vector & frames) override { + replaces_.fetch_add(1); + return inner_.replace_frames(fault_code, frames); + } + + tl::expected erase_frames(const std::string & fault_code) override { + erases_.fetch_add(1); + return inner_.erase_frames(fault_code); + } + + tl::expected, std::string> load_all() override { + return inner_.load_all(); + } + + /// Forget the writes the test itself made while seeding. + void reset_counts() { + replaces_.store(0); + erases_.store(0); + } + + int replaces() const { + return replaces_.load(); + } + int erases() const { + return erases_.load(); + } + + private: + InMemoryEntityFreezeFrameStore inner_; + std::atomic replaces_{0}; + std::atomic erases_{0}; +}; + +/// Route fetcher that serves a fixed level and records which entities it read, +/// so a test can prove the plant was NOT re-read for a fault that already has +/// a frame. +class CountingRouteFetcher { + public: + explicit CountingRouteFetcher(double level) : level_(level) { + } + + std::optional operator()(const std::string & entity_id) { + { + std::lock_guard lock(mutex_); + reads_[entity_id] += 1; + } + return json{{"connected", true}, {"items", json::array({{{"name", "level"}, {"value", level_}}})}}; + } + + int reads(const std::string & entity_id) { + std::lock_guard lock(mutex_); + auto it = reads_.find(entity_id); + return it == reads_.end() ? 0 : it->second; + } + + private: + double level_; + std::mutex mutex_; + std::map reads_; +}; + +} // namespace + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, CaptureWritesTheFrameThroughToTheStore) { + auto store = std::make_shared(); + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [this](const std::string & entity_id) -> DataProvider * { + return entity_id == "plc_app" ? provider_.get() : nullptr; + }, + nullptr, 256, nullptr, store); + + ASSERT_TRUE(publish_and_wait(capture, make_confirmed_event("PLC_PERSIST", {"plc_app"}))); + const auto served = capture.frames_for("PLC_PERSIST"); + ASSERT_EQ(served.size(), 1u); + + auto rows = store->load_all(); + ASSERT_TRUE(rows.has_value()); + ASSERT_EQ(rows->size(), 1u); + EXPECT_EQ((*rows)[0].fault_code, "PLC_PERSIST"); + EXPECT_EQ((*rows)[0].entity_id, "plc_app"); + EXPECT_EQ((*rows)[0].captured_at_ns, served[0].captured_at_ns); + EXPECT_EQ((*rows)[0].frame["values"], served[0].values); + EXPECT_EQ((*rows)[0].source, EntityFreezeFrameCapture::kSourceDataProvider); + EXPECT_EQ((*rows)[0].capture_origin, ""); // captured on the confirm edge +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, ReConfirmOverwritesTheStoredRow) { + auto store = std::make_shared(); + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [this](const std::string & entity_id) -> DataProvider * { + return entity_id == "plc_app" ? provider_.get() : nullptr; + }, + nullptr, 256, nullptr, store); + + ASSERT_TRUE(publish_and_wait(capture, make_confirmed_event("PLC_RECONFIRM", {"plc_app"}))); + + // The store must follow the map: today's semantics are one frame per fault, + // so a stale row would resurrect the previous occurrence's values on restart. + provider_->set_temperature(99.0); + const auto reconfirm = make_confirmed_event("PLC_RECONFIRM", {"plc_app"}); + const auto deadline = std::chrono::steady_clock::now() + 5s; + bool overwritten = false; + while (std::chrono::steady_clock::now() < deadline && !overwritten) { + publisher_->publish(reconfirm); + std::this_thread::sleep_for(20ms); + auto rows = store->load_all(); + overwritten = rows.has_value() && rows->size() == 1u && + std::abs((*rows)[0].frame["values"].value("temperature", 0.0) - 99.0) < 1e-9; + } + EXPECT_TRUE(overwritten); +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AReloadedFrameIsServedExactlyAsItWasCaptured) { + auto store = std::make_shared(); + json first_wire; + int64_t captured_at_ns = 0; + { + CountingRouteFetcher fetcher(41.0); + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&fetcher](const std::string & entity_id) { + return fetcher(entity_id); + }, + 256, nullptr, store); + ASSERT_TRUE(publish_and_wait(capture, make_confirmed_event("PLC_RESTART", {"route_plc_app"}))); + const auto frames = capture.frames_for("PLC_RESTART"); + ASSERT_EQ(frames.size(), 1u); + captured_at_ns = frames[0].captured_at_ns; + first_wire = FaultHandlers::merge_entity_freeze_frames(json{{"snapshots", json::array()}}, frames); + } + + // A second gateway life on the same store, with the plant now reading + // something else entirely: the served frame must still be the frozen one. + CountingRouteFetcher moved_on(999.0); + EntityFreezeFrameCapture restarted( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&moved_on](const std::string & entity_id) { + return moved_on(entity_id); + }, + 256, nullptr, store); + + const auto reloaded = restarted.frames_for("PLC_RESTART"); + ASSERT_EQ(reloaded.size(), 1u); + EXPECT_EQ(reloaded[0].captured_at_ns, captured_at_ns); + EXPECT_FALSE(reloaded[0].startup_catchup); + const auto second_wire = FaultHandlers::merge_entity_freeze_frames(json{{"snapshots", json::array()}}, reloaded); + EXPECT_EQ(second_wire, first_wire); // byte for byte, marker included + ASSERT_EQ(second_wire["snapshots"].size(), 1u); + EXPECT_FALSE(second_wire["snapshots"][0].contains("capture_origin")); +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, CatchUpSkipsAReloadedCodeAndFramesOneWithoutARow) { + auto store = std::make_shared(); + ASSERT_TRUE( + store->replace_frames("PLC_HAS_FRAME", {make_stored_row("PLC_HAS_FRAME", "route_stored_app", 4242)}).has_value()); + + CountingRouteFetcher fetcher(7.0); + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&fetcher](const std::string & entity_id) { + return fetcher(entity_id); + }, + 256, + [](const std::function &) -> std::vector { + return {{"PLC_HAS_FRAME", {"route_stored_app"}}, {"PLC_NO_FRAME", {"route_fresh_app"}}}; + }, + store); + + // Positive control on the same harness: a standing fault with no stored row + // still gets its startup frame, so an empty PLC_HAS_FRAME below would be a + // broken catch-up rather than a working skip. + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (capture.frames_for("PLC_NO_FRAME").empty() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(20ms); + } + const auto fresh = capture.frames_for("PLC_NO_FRAME"); + ASSERT_EQ(fresh.size(), 1u); + EXPECT_TRUE(fresh[0].startup_catchup); + + const auto stored = capture.frames_for("PLC_HAS_FRAME"); + ASSERT_EQ(stored.size(), 1u); + EXPECT_EQ(stored[0].captured_at_ns, 4242); // the frozen moment, not this start + EXPECT_FALSE(stored[0].startup_catchup); + EXPECT_EQ(fetcher.reads("route_stored_app"), 0); // the plant was never re-read for it +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, ReloadDropsAFaultTheManagerNoLongerHoldsAndKeepsAClearedOne) { + auto store = std::make_shared(); + ASSERT_TRUE(store->replace_frames("PLC_GONE", {make_stored_row("PLC_GONE", "route_a", 100)}).has_value()); + ASSERT_TRUE(store->replace_frames("PLC_CLEARED", {make_stored_row("PLC_CLEARED", "route_b", 200)}).has_value()); + + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + nullptr, 256, + [](const std::function &) -> std::vector { + return {}; // nothing standing: only the reload and the prune run + }, + store, + // The fault manager reports the cleared fault and knows nothing of the + // other: its own store was replaced under ours. + [](const std::function &) -> std::optional> { + return std::unordered_set{"PLC_CLEARED"}; + }); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (!capture.frames_for("PLC_GONE").empty() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(20ms); + } + EXPECT_TRUE(capture.frames_for("PLC_GONE").empty()); + // A cleared fault keeps its frame: the gateway's retention across a clear is + // exactly what persisting it is meant to preserve. + EXPECT_FALSE(capture.frames_for("PLC_CLEARED").empty()); + auto rows = store->load_all(); + ASSERT_TRUE(rows.has_value()); + ASSERT_EQ(rows->size(), 1u); + EXPECT_EQ((*rows)[0].fault_code, "PLC_CLEARED"); +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AnUnanswerableKnownFaultListerDropsNothing) { + // Absence assertion, controlled by the test above: the same seeding with a + // lister that CAN answer drops PLC_GONE, so "nothing dropped" here is the + // "could not tell" rule and not a prune that never runs. + auto store = std::make_shared(); + ASSERT_TRUE(store->replace_frames("PLC_GONE", {make_stored_row("PLC_GONE", "route_a", 100)}).has_value()); + ASSERT_TRUE(store->replace_frames("PLC_CLEARED", {make_stored_row("PLC_CLEARED", "route_b", 200)}).has_value()); + + std::atomic asked{false}; + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + nullptr, 256, + [](const std::function &) -> std::vector { + return {}; + }, + store, + [&asked](const std::function &) -> std::optional> { + asked.store(true); + return std::nullopt; // fault manager unreachable + }); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (!asked.load() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(20ms); + } + ASSERT_TRUE(asked.load()); + std::this_thread::sleep_for(300ms); // would be enough for a prune to land + EXPECT_FALSE(capture.frames_for("PLC_GONE").empty()); + EXPECT_FALSE(capture.frames_for("PLC_CLEARED").empty()); +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, TheRetainedFrameBoundCountsReloadedFrames) { + auto store = std::make_shared(); + ASSERT_TRUE(store->replace_frames("PLC_LOADED_A", {make_stored_row("PLC_LOADED_A", "route_a", 100)}).has_value()); + ASSERT_TRUE(store->replace_frames("PLC_LOADED_B", {make_stored_row("PLC_LOADED_B", "route_b", 200)}).has_value()); + + const auto standing = [](const std::function &) -> std::vector { + return {{"PLC_FRESH", {"route_fresh"}}}; + }; + CountingRouteFetcher fetcher(7.0); + const auto route = [&fetcher](const std::string & entity_id) { + return fetcher(entity_id); + }; + const auto no_provider = [](const std::string &) -> DataProvider * { + return nullptr; + }; + + { + // Bound of 2, already met by the two reloaded frames: catching PLC_FRESH up + // would FIFO-evict one of them, so it is refused instead. + EntityFreezeFrameCapture capped(node_.get(), *sub_exec_, no_provider, route, /*max_faults=*/2, standing, store); + std::this_thread::sleep_for(1s); // enough for an uncapped catch-up to land + EXPECT_TRUE(capped.frames_for("PLC_FRESH").empty()); + EXPECT_FALSE(capped.frames_for("PLC_LOADED_A").empty()); + EXPECT_FALSE(capped.frames_for("PLC_LOADED_B").empty()); + } + + // Positive control on the same harness: one more slot and the same catch-up + // frames it. + EntityFreezeFrameCapture roomy(node_.get(), *sub_exec_, no_provider, route, /*max_faults=*/3, standing, store); + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (roomy.frames_for("PLC_FRESH").empty() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(20ms); + } + EXPECT_FALSE(roomy.frames_for("PLC_FRESH").empty()); +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AnEvictedFaultLosesItsStoredRowToo) { + // Otherwise the bound holds only within a process: an evicted frame would + // come back on the next start and the file would grow without a limit. + auto store = std::make_shared(); + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [this](const std::string & entity_id) -> DataProvider * { + return entity_id == "plc_app" ? provider_.get() : nullptr; + }, + nullptr, /*max_faults=*/1, nullptr, store); + + ASSERT_TRUE(publish_and_wait(capture, make_confirmed_event("PLC_EVICT_FIRST", {"plc_app"}))); + ASSERT_TRUE(publish_and_wait(capture, make_confirmed_event("PLC_EVICT_SECOND", {"plc_app"}))); + + auto rows = store->load_all(); + ASSERT_TRUE(rows.has_value()); + ASSERT_EQ(rows->size(), 1u); + EXPECT_EQ((*rows)[0].fault_code, "PLC_EVICT_SECOND"); +} + +namespace { + +/// A capture whose store already holds one frame for PLC_REOCCUR, taken at +/// `stored_at_ns` with level 10.0, against a plant that now reads 99.0. The +/// standing lister reports the fault with `first_occurred_ns`, which is what +/// decides whether the stored frame belongs to the occurrence being served. +struct ReoccurrenceHarness { + std::shared_ptr store = std::make_shared(); + CountingRouteFetcher plant{99.0}; + static constexpr int64_t kStoredAtNs = 1'000'000'000'000'000'000; +}; + +/// Route fetcher whose entity never answers: the plugin-unreachable shape, and +/// the one the whole feature exists for (a restart while the link is down). +class UnreachableRouteFetcher { + public: + std::optional operator()(const std::string & entity_id) { + std::lock_guard lock(mutex_); + reads_[entity_id] += 1; + return std::nullopt; + } + + int reads(const std::string & entity_id) { + std::lock_guard lock(mutex_); + auto it = reads_.find(entity_id); + return it == reads_.end() ? 0 : it->second; + } + + private: + std::mutex mutex_; + std::map reads_; +}; + +/// Standing lister reporting one fault whose occurrence began after the frame +/// the store holds for it. +EntityFreezeFrameCapture::StandingFaultLister reoccurred_lister(const std::string & code, + const std::vector & sources) { + return [code, sources](const std::function &) { + EntityFreezeFrameCapture::StandingFault fault; + fault.fault_code = code; + fault.reporting_sources = sources; + fault.first_occurred_ns = ReoccurrenceHarness::kStoredAtNs + 60'000'000'000; + return std::vector{fault}; + }; +} + +/// Standing lister over an explicit list, so a test can fix the reply's order +/// and each fault's occurrence start independently. +EntityFreezeFrameCapture::StandingFaultLister +listed_faults(std::vector faults) { + return [faults](const std::function &) { + return faults; + }; +} + +EntityFreezeFrameCapture::StandingFault standing_fault(const std::string & code, const std::vector & srcs, + int64_t first_occurred_ns) { + EntityFreezeFrameCapture::StandingFault fault; + fault.fault_code = code; + fault.reporting_sources = srcs; + fault.first_occurred_ns = first_occurred_ns; + return fault; +} + +/// Route fetcher that answers for every entity except the named ones, so one +/// entity out of several can be the unreachable one. +class SelectiveRouteFetcher { + public: + SelectiveRouteFetcher(double level, std::set unreachable) + : level_(level), unreachable_(std::move(unreachable)) { + } + + std::optional operator()(const std::string & entity_id) { + { + std::lock_guard lock(mutex_); + reads_[entity_id] += 1; + } + if (unreachable_.count(entity_id) != 0) { + return std::nullopt; + } + return json{{"connected", true}, {"items", json::array({{{"name", "level"}, {"value", level_}}})}}; + } + + int reads(const std::string & entity_id) { + std::lock_guard lock(mutex_); + auto it = reads_.find(entity_id); + return it == reads_.end() ? 0 : it->second; + } + + private: + double level_; + std::set unreachable_; + std::mutex mutex_; + std::map reads_; +}; + +/// Route fetcher whose named entity fails its first N calls and answers after +/// that: a link that comes back between two reads inside one catch-up. +class FlakyRouteFetcher { + public: + FlakyRouteFetcher(double level, std::string flaky, int failures) + : level_(level), flaky_(std::move(flaky)), failures_left_(failures) { + } + + std::optional operator()(const std::string & entity_id) { + std::lock_guard lock(mutex_); + reads_[entity_id] += 1; + if (entity_id == flaky_ && failures_left_ > 0) { + --failures_left_; + return std::nullopt; + } + return json{{"connected", true}, {"items", json::array({{{"name", "level"}, {"value", level_}}})}}; + } + + int reads(const std::string & entity_id) { + std::lock_guard lock(mutex_); + auto it = reads_.find(entity_id); + return it == reads_.end() ? 0 : it->second; + } + + private: + double level_; + std::string flaky_; + int failures_left_; + std::mutex mutex_; + std::map reads_; +}; + +/// Fault codes present in the store, sorted, for a whole-store assertion. +std::vector stored_codes(const std::shared_ptr & store) { + auto rows = store->load_all(); + std::vector codes; + if (rows) { + for (const auto & row : *rows) { + codes.push_back(row.fault_code); + } + } + std::sort(codes.begin(), codes.end()); + return codes; +} + +// The bound probe both tests below run. Two slots, one live reloaded frame +// (PLC_KEEP, the older of the two so it is the FIFO front), one stale reloaded +// frame (PLC_STALE) and one standing fault with no frame at all (PLC_NEW). +constexpr int64_t kKeepAtNs = ReoccurrenceHarness::kStoredAtNs; +constexpr int64_t kStaleAtNs = ReoccurrenceHarness::kStoredAtNs + 10'000'000'000; + +std::vector bound_probe_standing() { + return {standing_fault("PLC_NEW", {"route_new"}, kStaleAtNs + 30'000'000'000), + standing_fault("PLC_STALE", {"route_stale"}, kStaleAtNs + 60'000'000'000), // after its frame, so stale + standing_fault("PLC_KEEP", {"route_keep"}, kKeepAtNs - 60'000'000'000)}; // before its frame, so live +} + +void seed_bound_probe(const std::shared_ptr & store) { + ASSERT_TRUE( + store->replace_frames("PLC_KEEP", {make_stored_row("PLC_KEEP", "route_keep", kKeepAtNs, 10.0)}).has_value()); + ASSERT_TRUE( + store->replace_frames("PLC_STALE", {make_stored_row("PLC_STALE", "route_stale", kStaleAtNs, 11.0)}).has_value()); +} + +} // namespace + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AReloadedFrameFromAnEarlierOccurrenceIsReReadAndMarkedStartup) { + // The fault cleared and confirmed again while the gateway was down, so the + // stored frame holds the PREVIOUS incident's values. Serving it unmarked + // would present last week's numbers as this occurrence's. + ReoccurrenceHarness h; + ASSERT_TRUE( + h.store->replace_frames("PLC_REOCCUR", {make_stored_row("PLC_REOCCUR", "route_stored_app", h.kStoredAtNs, 10.0)}) + .has_value()); + h.store->reset_counts(); + + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&h](const std::string & entity_id) { + return h.plant(entity_id); + }, + 256, reoccurred_lister("PLC_REOCCUR", {"route_stored_app"}), h.store); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (std::chrono::steady_clock::now() < deadline) { + const auto now = capture.frames_for("PLC_REOCCUR"); + if (!now.empty() && std::abs(now[0].values.value("level", 0.0) - 99.0) < 1e-9) { + break; + } + std::this_thread::sleep_for(20ms); + } + const auto served = capture.frames_for("PLC_REOCCUR"); + ASSERT_EQ(served.size(), 1u); + EXPECT_DOUBLE_EQ(served[0].values.value("level", 0.0), 99.0); // this occurrence, not the last one + EXPECT_TRUE(served[0].startup_catchup); // read at start, so it says so + EXPECT_GT(served[0].captured_at_ns, ReoccurrenceHarness::kStoredAtNs); + + auto rows = h.store->load_all(); + ASSERT_TRUE(rows.has_value()); + ASSERT_EQ(rows->size(), 1u); + EXPECT_DOUBLE_EQ((*rows)[0].frame["values"].value("level", 0.0), 99.0); // replaced on disk too + EXPECT_EQ((*rows)[0].capture_origin, "startup"); + + // The order, not just the outcome: the plant is read first and the row is + // then swapped by ONE write. Erasing first and re-taking afterwards leaves + // the fault with nothing whenever the entity cannot answer, which is exactly + // the case this feature is for. + EXPECT_EQ(h.store->erases(), 0); + EXPECT_EQ(h.store->replaces(), 1); + EXPECT_EQ(h.plant.reads("route_stored_app"), 1); +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AStaleFrameWhoseEntityCannotBeReReadIsDroppedWithAWarning) { + // The headline case: the gateway restarts while the PLC link is down, and the + // fault re-confirmed in the meantime. The stored frame is from the previous + // occurrence so it must not be served, and the re-read cannot replace it, so + // the fault ends with no frame. That is a real loss of evidence and the log + // is the only place the operator can learn about it. + auto store = std::make_shared(); + UnreachableRouteFetcher plant; + ASSERT_TRUE(store + ->replace_frames("PLC_LINK_DOWN", {make_stored_row("PLC_LINK_DOWN", "route_stored_app", + ReoccurrenceHarness::kStoredAtNs, 10.0)}) + .has_value()); + + testing::internal::CaptureStderr(); + { + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&plant](const std::string & entity_id) { + return plant(entity_id); + }, + 256, reoccurred_lister("PLC_LINK_DOWN", {"route_stored_app"}), store); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (std::chrono::steady_clock::now() < deadline && plant.reads("route_stored_app") == 0) { + std::this_thread::sleep_for(20ms); + } + std::this_thread::sleep_for(300ms); // let the drop that follows the failed read land + EXPECT_GE(plant.reads("route_stored_app"), 1); // the re-read WAS attempted + EXPECT_TRUE(capture.frames_for("PLC_LINK_DOWN").empty()); + } + const auto logs = testing::internal::GetCapturedStderr(); + + auto rows = store->load_all(); + ASSERT_TRUE(rows.has_value()); + EXPECT_TRUE(rows->empty()); // and the row is gone, not left to be served + EXPECT_NE(logs.find("PLC_LINK_DOWN"), std::string::npos) << logs; + EXPECT_NE(logs.find("route_stored_app"), std::string::npos) << logs; + EXPECT_NE(logs.find("no freeze-frame"), std::string::npos) << logs; +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AStaleFrameWhoseFaultNamesNoEntityIsDroppedWithAWarning) { + // The two eligibility tests have to agree. A standing fault with no reporting + // sources is stale to the comparison and unreadable to the catch-up, so it is + // a re-read that cannot even be attempted, and it gets the same treatment and + // the same line rather than disappearing quietly. + auto store = std::make_shared(); + CountingRouteFetcher plant{99.0}; + ASSERT_TRUE(store + ->replace_frames("PLC_NO_ENTITY", {make_stored_row("PLC_NO_ENTITY", "route_stored_app", + ReoccurrenceHarness::kStoredAtNs, 10.0)}) + .has_value()); + + testing::internal::CaptureStderr(); + { + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&plant](const std::string & entity_id) { + return plant(entity_id); + }, + 256, reoccurred_lister("PLC_NO_ENTITY", {}), store); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (std::chrono::steady_clock::now() < deadline && !capture.frames_for("PLC_NO_ENTITY").empty()) { + std::this_thread::sleep_for(20ms); + } + EXPECT_TRUE(capture.frames_for("PLC_NO_ENTITY").empty()); + EXPECT_EQ(plant.reads("route_stored_app"), 0); // nothing to read, and none was invented + } + const auto logs = testing::internal::GetCapturedStderr(); + + auto rows = store->load_all(); + ASSERT_TRUE(rows.has_value()); + EXPECT_TRUE(rows->empty()); + EXPECT_NE(logs.find("PLC_NO_ENTITY"), std::string::npos) << logs; + EXPECT_NE(logs.find("no entity to read"), std::string::npos) << logs; +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AReloadedFrameFromTheSameOccurrenceIsKeptUnmarked) { + // Control for the test above on the same harness: the fault never cleared, so + // its first_occurred predates the frame and the stored values are the ones + // this occurrence froze. Re-reading the plant here is the whole defect. + ReoccurrenceHarness h; + ASSERT_TRUE( + h.store->replace_frames("PLC_REOCCUR", {make_stored_row("PLC_REOCCUR", "route_stored_app", h.kStoredAtNs, 10.0)}) + .has_value()); + + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&h](const std::string & entity_id) { + return h.plant(entity_id); + }, + 256, + [](const std::function &) -> std::vector { + EntityFreezeFrameCapture::StandingFault fault; + fault.fault_code = "PLC_REOCCUR"; + fault.reporting_sources = {"route_stored_app"}; + fault.first_occurred_ns = ReoccurrenceHarness::kStoredAtNs - 60'000'000'000; // a minute before the frame + return std::vector{fault}; + }, + h.store); + + std::this_thread::sleep_for(1s); // enough for a catch-up read to land + const auto served = capture.frames_for("PLC_REOCCUR"); + ASSERT_EQ(served.size(), 1u); + EXPECT_DOUBLE_EQ(served[0].values.value("level", 0.0), 10.0); + EXPECT_EQ(served[0].captured_at_ns, ReoccurrenceHarness::kStoredAtNs); + EXPECT_FALSE(served[0].startup_catchup); + EXPECT_EQ(h.plant.reads("route_stored_app"), 0); // the plant was never re-read + + auto rows = h.store->load_all(); + ASSERT_TRUE(rows.has_value()); + ASSERT_EQ(rows->size(), 1u); + EXPECT_DOUBLE_EQ((*rows)[0].frame["values"].value("level", 0.0), 10.0); // row untouched + EXPECT_EQ((*rows)[0].captured_at_ns, ReoccurrenceHarness::kStoredAtNs); +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AStaleReReadAtTheBoundNeverCostsALiveFrame) { + // Two slots, both taken by reloaded frames, and a third standing fault with + // none. The stale one owns its slot and its replacement cannot exceed the + // bound, so it must be re-read in place. The bare one has no slot, so it goes + // unframed - and must not be let in against an occupancy that counts the + // stale code as absent, because the FIFO would then evict PLC_KEEP, which is + // a live frame for a fault that is still standing. + auto store = std::make_shared(); + seed_bound_probe(store); + SelectiveRouteFetcher plant(99.0, {}); + + testing::internal::CaptureStderr(); + { + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&plant](const std::string & entity_id) { + return plant(entity_id); + }, + /*max_faults=*/2, listed_faults(bound_probe_standing()), store); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (std::chrono::steady_clock::now() < deadline && plant.reads("route_stale") == 0) { + std::this_thread::sleep_for(20ms); + } + std::this_thread::sleep_for(500ms); // enough for an unbounded pass to admit PLC_NEW + + const auto keep = capture.frames_for("PLC_KEEP"); + ASSERT_EQ(keep.size(), 1u) << "the live reloaded frame was evicted"; + EXPECT_DOUBLE_EQ(keep[0].values.value("level", 0.0), 10.0); + EXPECT_EQ(keep[0].captured_at_ns, kKeepAtNs); + EXPECT_FALSE(keep[0].startup_catchup); + + const auto stale = capture.frames_for("PLC_STALE"); + ASSERT_EQ(stale.size(), 1u); + EXPECT_DOUBLE_EQ(stale[0].values.value("level", 0.0), 99.0); // re-read in place + EXPECT_TRUE(stale[0].startup_catchup); + EXPECT_GT(stale[0].captured_at_ns, kStaleAtNs); + + EXPECT_TRUE(capture.frames_for("PLC_NEW").empty()); // no slot for it + + EXPECT_EQ(plant.reads("route_keep"), 0); // a live frame is never re-read + EXPECT_EQ(plant.reads("route_stale"), 1); // the stale one is, exactly once + EXPECT_EQ(plant.reads("route_new"), 0); // refused before the plugin was touched + } + const auto logs = testing::internal::GetCapturedStderr(); + + EXPECT_EQ(stored_codes(store), (std::vector{"PLC_KEEP", "PLC_STALE"})); + EXPECT_NE(logs.find("PLC_NEW"), std::string::npos) << logs; // the truncation names it +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AFailedStaleReReadFreesItsSlotForAFaultWithNoFrame) { + // Same probe, but the stale code's entity cannot answer. Its row is dropped, + // which genuinely frees a slot, so the bare fault now fits - and PLC_KEEP is + // still not the one that pays for it. + auto store = std::make_shared(); + seed_bound_probe(store); + SelectiveRouteFetcher plant(99.0, {"route_stale"}); + + testing::internal::CaptureStderr(); + { + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&plant](const std::string & entity_id) { + return plant(entity_id); + }, + /*max_faults=*/2, listed_faults(bound_probe_standing()), store); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (std::chrono::steady_clock::now() < deadline && capture.frames_for("PLC_NEW").empty()) { + std::this_thread::sleep_for(20ms); + } + + const auto keep = capture.frames_for("PLC_KEEP"); + ASSERT_EQ(keep.size(), 1u) << "the live reloaded frame was evicted"; + EXPECT_EQ(keep[0].captured_at_ns, kKeepAtNs); + EXPECT_FALSE(keep[0].startup_catchup); + + EXPECT_TRUE(capture.frames_for("PLC_STALE").empty()); // dropped, not served + + const auto fresh = capture.frames_for("PLC_NEW"); + ASSERT_EQ(fresh.size(), 1u) << "the freed slot was not reused"; + EXPECT_TRUE(fresh[0].startup_catchup); + + EXPECT_EQ(plant.reads("route_keep"), 0); + EXPECT_EQ(plant.reads("route_stale"), 1); // asked once before being dropped, never twice + EXPECT_EQ(plant.reads("route_new"), 1); + } + const auto logs = testing::internal::GetCapturedStderr(); + + EXPECT_EQ(stored_codes(store), (std::vector{"PLC_KEEP", "PLC_NEW"})); + EXPECT_NE(logs.find("PLC_STALE"), std::string::npos) << logs; + EXPECT_NE(logs.find("route_stale"), std::string::npos) << logs; + EXPECT_NE(logs.find("no freeze-frame"), std::string::npos) << logs; +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AStaleCodeIsAskedOnceEvenWithASlotToSpare) { + // Three slots for three faults, so nothing competes and the bound never + // speaks. A stale code whose entity cannot answer is dropped, which leaves a + // fault with no frame and a free slot: exactly the shape in which a second + // pass would go back and block on the same dead entity all over again. + auto store = std::make_shared(); + seed_bound_probe(store); + SelectiveRouteFetcher plant(99.0, {"route_stale"}); + + testing::internal::CaptureStderr(); + { + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&plant](const std::string & entity_id) { + return plant(entity_id); + }, + /*max_faults=*/3, listed_faults(bound_probe_standing()), store); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (std::chrono::steady_clock::now() < deadline && capture.frames_for("PLC_NEW").empty()) { + std::this_thread::sleep_for(20ms); + } + std::this_thread::sleep_for(500ms); // enough for a second pass at route_stale to land + + EXPECT_EQ(plant.reads("route_stale"), 1); // one blocking read per catch-up, whatever the capacity + EXPECT_EQ(plant.reads("route_keep"), 0); + EXPECT_EQ(plant.reads("route_new"), 1); + EXPECT_TRUE(capture.frames_for("PLC_STALE").empty()); // dropped, and it stays dropped + ASSERT_EQ(capture.frames_for("PLC_KEEP").size(), 1u); + EXPECT_EQ(capture.frames_for("PLC_KEEP")[0].captured_at_ns, kKeepAtNs); + ASSERT_EQ(capture.frames_for("PLC_NEW").size(), 1u); + } + const auto logs = testing::internal::GetCapturedStderr(); + + EXPECT_EQ(stored_codes(store), (std::vector{"PLC_KEEP", "PLC_NEW"})); + EXPECT_NE(logs.find("PLC_STALE"), std::string::npos) << logs; +} + +/// @verifies REQ_INTEROP_088 +TEST_F(EntityFreezeFrameCaptureTest, AStaleCodeThatWouldAnswerOnASecondAskIsStillOnlyAskedOnce) { + // The entity fails once and would answer if asked again. Asking again is what + // must not happen: the warning has already told the operator the frame was + // discarded and this occurrence has none, so a frame appearing anyway would + // make the log a lie, the summary would count a discard that did not stick, + // and the slot freed by the drop would go back to the code that just lost it + // instead of to the fault still waiting for one. + auto store = std::make_shared(); + seed_bound_probe(store); + FlakyRouteFetcher plant(99.0, "route_stale", /*failures=*/1); + // STALE first in the reply, so nothing else can reach the bound before it. + auto standing = bound_probe_standing(); + std::swap(standing[0], standing[1]); + + testing::internal::CaptureStderr(); + { + EntityFreezeFrameCapture capture( + node_.get(), *sub_exec_, + [](const std::string &) -> DataProvider * { + return nullptr; + }, + [&plant](const std::string & entity_id) { + return plant(entity_id); + }, + /*max_faults=*/2, listed_faults(standing), store); + + const auto deadline = std::chrono::steady_clock::now() + 15s; + while (std::chrono::steady_clock::now() < deadline && capture.frames_for("PLC_NEW").empty()) { + std::this_thread::sleep_for(20ms); + } + std::this_thread::sleep_for(500ms); + + EXPECT_EQ(plant.reads("route_stale"), 1); + EXPECT_TRUE(capture.frames_for("PLC_STALE").empty()) << "the discarded frame came back"; + ASSERT_EQ(capture.frames_for("PLC_KEEP").size(), 1u); + EXPECT_EQ(capture.frames_for("PLC_KEEP")[0].captured_at_ns, kKeepAtNs); + ASSERT_EQ(capture.frames_for("PLC_NEW").size(), 1u) << "the freed slot did not reach the waiting fault"; + } + const auto logs = testing::internal::GetCapturedStderr(); + + EXPECT_EQ(stored_codes(store), (std::vector{"PLC_KEEP", "PLC_NEW"})); + EXPECT_NE(logs.find("PLC_STALE"), std::string::npos) << logs; + EXPECT_NE(logs.find("no freeze-frame"), std::string::npos) << logs; + // The summary has to describe what actually happened, not what was attempted. + EXPECT_NE(logs.find("0 reloaded frame(s) from an earlier occurrence re-read, 1 discarded"), std::string::npos) + << logs; +} + TEST(ContentHasLiveData, GatesOnItemsNotOnTheLinkFlag) { using Capture = EntityFreezeFrameCapture; EXPECT_TRUE(Capture::content_has_live_data( @@ -757,6 +1657,54 @@ TEST(StandingFaultsFromListReply, ParsesWellFormedReply) { EXPECT_TRUE((*standing)[1].reporting_sources.empty()); } +TEST(StandingFaultsFromListReply, FirstOccurredIsReadInSecondsAndKeptInNanoseconds) { + // The wire carries seconds (fault_msg_conversions), the comparison against a + // frame's captured_at_ns needs nanoseconds. Anything that is not a positive + // number leaves 0, which the caller reads as "cannot tell" and never lets + // cost a stored frame. + const json data = { + {"faults", + json::array({json{{"fault_code", "SECONDS"}, + {"reporting_sources", json::array({"a"})}, + {"first_occurred", 1788948705.5}}, + json{{"fault_code", "ABSENT"}, {"reporting_sources", json::array({"a"})}}, + json{{"fault_code", "NOT_A_NUMBER"}, + {"reporting_sources", json::array({"a"})}, + {"first_occurred", "yesterday"}}, + json{{"fault_code", "ZERO"}, {"reporting_sources", json::array({"a"})}, {"first_occurred", 0}}, + // Past what int64 holds once multiplied out, so the cast + // itself is undefined. What the platform then hands back + // decides the behaviour, which is the whole problem. On + // x86-64 it is INT64_MIN, which the non-positive check in + // stale_reloaded_codes happens to reject, so the frame + // survives by luck. A saturating target hands back a large + // POSITIVE value instead, which passes that check and sits + // above every captured_at, so the fault reads as re-occurred + // and a good frame is discarded. The range guard is what + // makes the outcome the same on both. + json{{"fault_code", "HUGE"}, {"reporting_sources", json::array({"a"})}, {"first_occurred", 1e19}}, + json{{"fault_code", "INFINITE"}, + {"reporting_sources", json::array({"a"})}, + {"first_occurred", std::numeric_limits::infinity()}}, + json{{"fault_code", "NAN_SECONDS"}, + {"reporting_sources", json::array({"a"})}, + {"first_occurred", std::numeric_limits::quiet_NaN()}}})}}; + const auto standing = EntityFreezeFrameCapture::standing_faults_from_list_reply(data); + ASSERT_TRUE(standing.has_value()); + ASSERT_EQ(standing->size(), 7u); + EXPECT_EQ((*standing)[0].fault_code, "SECONDS"); + EXPECT_EQ((*standing)[0].first_occurred_ns, 1788948705500000000); + EXPECT_EQ((*standing)[1].first_occurred_ns, 0); // absent + EXPECT_EQ((*standing)[2].first_occurred_ns, 0); // not a number + EXPECT_EQ((*standing)[3].first_occurred_ns, 0); // zero + EXPECT_EQ((*standing)[4].fault_code, "HUGE"); + EXPECT_EQ((*standing)[4].first_occurred_ns, 0); + EXPECT_EQ((*standing)[5].fault_code, "INFINITE"); + EXPECT_EQ((*standing)[5].first_occurred_ns, 0); + EXPECT_EQ((*standing)[6].fault_code, "NAN_SECONDS"); + EXPECT_EQ((*standing)[6].first_occurred_ns, 0); +} + TEST(StandingFaultsFromListReply, RepliesNotShapedLikeListFaultsYieldNullopt) { // nullopt (vs empty vector) is what lets the caller warn on a malformed or // renamed reply instead of silently disabling the catch-up. diff --git a/src/ros2_medkit_gateway/test/test_entity_freeze_frame_store.cpp b/src/ros2_medkit_gateway/test/test_entity_freeze_frame_store.cpp new file mode 100644 index 000000000..00e564137 --- /dev/null +++ b/src/ros2_medkit_gateway/test/test_entity_freeze_frame_store.cpp @@ -0,0 +1,225 @@ +// 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. + +#include + +#include + +#include +#include +#include +#include + +#include "ros2_medkit_gateway/core/entity_freeze_frame_store.hpp" +#include "ros2_medkit_gateway/core/sqlite_entity_freeze_frame_store.hpp" + +using json = nlohmann::json; +using ros2_medkit_gateway::EntityFreezeFrameStore; +using ros2_medkit_gateway::InMemoryEntityFreezeFrameStore; +using ros2_medkit_gateway::SqliteEntityFreezeFrameStore; +using ros2_medkit_gateway::StoredEntityFreezeFrame; + +namespace { + +StoredEntityFreezeFrame make_row(const std::string & fault_code, const std::string & entity_id, int64_t captured_at_ns, + json frame = json{{"values", {{"temperature", 42.5}, {"pressure", 3.2}}}, + {"connected", false}, + {"source_timestamp", "2026-09-08T17:51:40.387Z"}}) { + StoredEntityFreezeFrame row; + row.fault_code = fault_code; + row.entity_id = entity_id; + row.frame = std::move(frame); + row.captured_at_ns = captured_at_ns; + row.source = "plugin_x_plc_data_route"; + row.capture_origin = ""; + return row; +} + +/// Both backends must behave identically: the in-memory one is what tests and +/// a path-less gateway get, and a difference between them would only show up +/// on the box. +enum class Backend { InMemory, Sqlite }; + +class EntityFreezeFrameStoreTest : public ::testing::TestWithParam { + protected: + void SetUp() override { + db_path_ = std::filesystem::temp_directory_path() / + ("test_entity_freeze_frame_store_" + std::to_string(::getpid()) + ".db"); + std::filesystem::remove(db_path_); + store_ = open(); + } + + void TearDown() override { + store_.reset(); + std::filesystem::remove(db_path_); + } + + std::unique_ptr open() { + if (GetParam() == Backend::InMemory) { + return std::make_unique(); + } + return std::make_unique(db_path_.string()); + } + + std::filesystem::path db_path_; + std::unique_ptr store_; +}; + +} // namespace + +/// @verifies REQ_INTEROP_088 +TEST_P(EntityFreezeFrameStoreTest, RoundTripsEveryFieldOfARow) { + auto row = make_row("JAM_INFEED", "plc_app", 1757353900387000000); + row.capture_origin = "startup"; + ASSERT_TRUE(store_->replace_frames("JAM_INFEED", {row}).has_value()); + + auto loaded = store_->load_all(); + ASSERT_TRUE(loaded.has_value()); + ASSERT_EQ(loaded->size(), 1u); + const auto & got = (*loaded)[0]; + EXPECT_EQ(got.fault_code, "JAM_INFEED"); + EXPECT_EQ(got.entity_id, "plc_app"); + EXPECT_EQ(got.frame, row.frame); + EXPECT_EQ(got.captured_at_ns, 1757353900387000000); // nanoseconds need the full 64 bits + EXPECT_EQ(got.source, "plugin_x_plc_data_route"); + EXPECT_EQ(got.capture_origin, "startup"); +} + +/// @verifies REQ_INTEROP_088 +TEST_P(EntityFreezeFrameStoreTest, ReplaceDropsEntitiesTheNewCaptureNoLongerReports) { + ASSERT_TRUE(store_ + ->replace_frames("JAM_INFEED", + {make_row("JAM_INFEED", "plc_app", 10), make_row("JAM_INFEED", "second_app", 20)}) + .has_value()); + // A re-confirm that only frames one entity must not leave the other's row + // behind: the served frames are replaced as a unit, so the rows are too. + ASSERT_TRUE(store_->replace_frames("JAM_INFEED", {make_row("JAM_INFEED", "plc_app", 30)}).has_value()); + + auto loaded = store_->load_all(); + ASSERT_TRUE(loaded.has_value()); + ASSERT_EQ(loaded->size(), 1u); + EXPECT_EQ((*loaded)[0].entity_id, "plc_app"); + EXPECT_EQ((*loaded)[0].captured_at_ns, 30); +} + +/// @verifies REQ_INTEROP_088 +TEST_P(EntityFreezeFrameStoreTest, OneCodePerFaultRowsAreKeptApart) { + ASSERT_TRUE(store_->replace_frames("JAM_INFEED", {make_row("JAM_INFEED", "plc_app", 10)}).has_value()); + ASSERT_TRUE(store_->replace_frames("SAFETY_CURTAIN", {make_row("SAFETY_CURTAIN", "plc_app", 20)}).has_value()); + + ASSERT_TRUE(store_->erase_frames("JAM_INFEED").has_value()); + + auto loaded = store_->load_all(); + ASSERT_TRUE(loaded.has_value()); + ASSERT_EQ(loaded->size(), 1u); + EXPECT_EQ((*loaded)[0].fault_code, "SAFETY_CURTAIN"); +} + +/// @verifies REQ_INTEROP_088 +TEST_P(EntityFreezeFrameStoreTest, LoadAllServesOldestCaptureFirst) { + // The caller keeps the newest codes when the store holds more than its + // retained-frame bound, so the order is part of the contract, not a detail. + ASSERT_TRUE(store_->replace_frames("NEWEST", {make_row("NEWEST", "plc_app", 300)}).has_value()); + ASSERT_TRUE(store_->replace_frames("OLDEST", {make_row("OLDEST", "plc_app", 100)}).has_value()); + ASSERT_TRUE(store_->replace_frames("MIDDLE", {make_row("MIDDLE", "plc_app", 200)}).has_value()); + + auto loaded = store_->load_all(); + ASSERT_TRUE(loaded.has_value()); + ASSERT_EQ(loaded->size(), 3u); + EXPECT_EQ((*loaded)[0].fault_code, "OLDEST"); + EXPECT_EQ((*loaded)[1].fault_code, "MIDDLE"); + EXPECT_EQ((*loaded)[2].fault_code, "NEWEST"); +} + +/// @verifies REQ_INTEROP_088 +TEST_P(EntityFreezeFrameStoreTest, ErasingACodeWithNoRowsIsNotAnError) { + // The caller evicts by fault code and does not track what reached the file. + EXPECT_TRUE(store_->erase_frames("NEVER_STORED").has_value()); +} + +/// @verifies REQ_INTEROP_088 +TEST_P(EntityFreezeFrameStoreTest, AnEmptyReplaceLeavesNoRows) { + ASSERT_TRUE(store_->replace_frames("JAM_INFEED", {make_row("JAM_INFEED", "plc_app", 10)}).has_value()); + ASSERT_TRUE(store_->replace_frames("JAM_INFEED", {}).has_value()); + + auto loaded = store_->load_all(); + ASSERT_TRUE(loaded.has_value()); + EXPECT_TRUE(loaded->empty()); +} + +INSTANTIATE_TEST_SUITE_P(Backends, EntityFreezeFrameStoreTest, ::testing::Values(Backend::InMemory, Backend::Sqlite), + [](const ::testing::TestParamInfo & param_info) { + return param_info.param == Backend::InMemory ? "InMemory" : "Sqlite"; + }); + +// =========================================================================== +// SQLite only: the point of the file is that it outlives the process. +// =========================================================================== + +/// @verifies REQ_INTEROP_088 +TEST(SqliteEntityFreezeFrameStoreFile, RowsSurviveReopen) { + const auto path = std::filesystem::temp_directory_path() / + ("test_entity_freeze_frame_reopen_" + std::to_string(::getpid()) + ".db"); + std::filesystem::remove(path); + + { + SqliteEntityFreezeFrameStore store(path.string()); + ASSERT_TRUE( + store.replace_frames("JAM_INFEED", {make_row("JAM_INFEED", "plc_app", 1757353900387000000)}).has_value()); + } + + SqliteEntityFreezeFrameStore reopened(path.string()); + auto loaded = reopened.load_all(); + ASSERT_TRUE(loaded.has_value()); + ASSERT_EQ(loaded->size(), 1u); + EXPECT_EQ((*loaded)[0].fault_code, "JAM_INFEED"); + EXPECT_EQ((*loaded)[0].captured_at_ns, 1757353900387000000); + EXPECT_EQ((*loaded)[0].frame["values"]["temperature"], 42.5); + EXPECT_EQ((*loaded)[0].frame["connected"], false); + + std::filesystem::remove(path); +} + +/// @verifies REQ_INTEROP_088 +TEST(SqliteEntityFreezeFrameStoreFile, AnUnreadableRowCostsOnlyItself) { + const auto path = std::filesystem::temp_directory_path() / + ("test_entity_freeze_frame_corrupt_" + std::to_string(::getpid()) + ".db"); + std::filesystem::remove(path); + + { + SqliteEntityFreezeFrameStore store(path.string()); + ASSERT_TRUE(store.replace_frames("GOOD", {make_row("GOOD", "plc_app", 20)}).has_value()); + } + // Hand-edit one blob into something that is not JSON, as a half-written file + // or a fat-fingered sqlite3 session would. + { + sqlite3 * db = nullptr; + ASSERT_EQ(sqlite3_open(path.c_str(), &db), SQLITE_OK); + ASSERT_EQ(sqlite3_exec(db, + "INSERT INTO entity_freeze_frames " + "(fault_code, entity_id, frame, captured_at_ns, source, capture_origin) " + "VALUES ('BROKEN','plc_app','{not json',10,'','')", + nullptr, nullptr, nullptr), + SQLITE_OK); + sqlite3_close(db); + } + + SqliteEntityFreezeFrameStore reopened(path.string()); + auto loaded = reopened.load_all(); + ASSERT_TRUE(loaded.has_value()); + ASSERT_EQ(loaded->size(), 1u); + EXPECT_EQ((*loaded)[0].fault_code, "GOOD"); + + std::filesystem::remove(path); +} diff --git a/src/ros2_medkit_plugins/ros2_medkit_opcua/README.md b/src/ros2_medkit_plugins/ros2_medkit_opcua/README.md index 7b6abdde0..e3f142d3f 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_opcua/README.md +++ b/src/ros2_medkit_plugins/ros2_medkit_opcua/README.md @@ -871,10 +871,28 @@ curl -s http://localhost:8080/api/v1/apps/tank_process/x-plc-data | jq . # Automated tests (16 assertions) bash scripts/run_integration_tests.sh -# Stop +# Stop (keeps the gateway's state) bash scripts/stop.sh ``` +`start.sh` mounts the named volume `ros2-medkit-opcua-state` at +`/var/lib/ros2_medkit`, so the gateway's state survives `stop.sh` and the next +`start.sh`: the entity freeze frames, `faults.db` and the rosbags. That is what +lets a fault raised before the stop still serve the values frozen when it +confirmed, with its original `captured_at` and no `x-medkit.capture_origin`, +rather than a fresh read of the PLC as it is after the restart. `stop.sh` +removes the container, so state left in the container's writable layer would +not survive it. + +To start from a clean slate, purge the volume: + +```bash +bash scripts/stop.sh +docker volume rm ros2-medkit-opcua-state +``` + +Set `OPCUA_DEMO_STATE_VOLUME` to use a different volume name. + A separate scenario covers the config-less discovery start-up race, which the suite above cannot see because it pins `OPCUA_ENDPOINT_URL` and so short-circuits discovery. It starts the gateway before any server, with diff --git a/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/gateway_params.yaml b/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/gateway_params.yaml index a331166cb..f03bc668a 100644 --- a/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/gateway_params.yaml +++ b/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/gateway_params.yaml @@ -12,3 +12,14 @@ ros2_medkit_gateway: allowed_origins: ["*"] plugins: ["opcua"] plugins.opcua.poll_interval_ms: 1000 + # Freeze-frames for the PLC entities are written next to the rest of the + # gateway's state, so a restart serves the values frozen when the fault + # confirmed instead of re-reading the PLC as it is now. scripts/start.sh + # mounts the named volume "ros2-medkit-opcua-state" here, which is what + # carries the frames across scripts/stop.sh too: that script removes the + # container, so the writable layer would not survive it. The test scripts + # start their own containers without a volume and keep the frames inside + # the container, which is all a single-run suite needs. + entity_freeze_frame: + storage: + path: "/var/lib/ros2_medkit/entity_freeze_frames.db" diff --git a/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/start.sh b/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/start.sh index f80d00fbc..bac6fc6ca 100755 --- a/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/start.sh +++ b/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/start.sh @@ -2,8 +2,14 @@ # Start OpenPLC + medkit gateway for manual testing. # Usage: from the ros2_medkit repo root, run # bash src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/start.sh +# +# The gateway's state (entity freeze frames, faults.db, rosbags) is kept on the +# named volume below, so it survives stop.sh and a later start.sh. Purge it with +# docker volume rm ros2-medkit-opcua-state set -eo pipefail +STATE_VOLUME="${OPCUA_DEMO_STATE_VOLUME:-ros2-medkit-opcua-state}" + SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" DOCKER_DIR="$(dirname "$SCRIPT_DIR")" PLUGIN_DIR="$(dirname "$DOCKER_DIR")" @@ -23,6 +29,12 @@ echo "" echo "=== Starting containers ===" docker rm -f openplc gateway 2>/dev/null || true docker network create plc-demo 2>/dev/null || true +# The gateway's state goes on a named volume rather than the container's +# writable layer, because stop.sh removes the container. Without this a +# freeze-frame captured when the alarm confirmed would be destroyed by the only +# stop procedure the demo ships, and the next start would re-read the PLC as it +# is then instead of serving the values frozen at fault time. +docker volume create "$STATE_VOLUME" >/dev/null docker run -d --name openplc --network plc-demo -p 4840:4840 openplc-tank echo "OpenPLC starting..." @@ -35,6 +47,7 @@ for _ in $(seq 1 45); do done docker run -d --name gateway --network plc-demo -p 8080:8080 \ + -v "$STATE_VOLUME":/var/lib/ros2_medkit \ -e ROS_DOMAIN_ID=60 \ -e OPCUA_ENDPOINT_URL="opc.tcp://openplc:4840/openplc/opcua" \ -e OPCUA_NODE_MAP_PATH="/config/tank_nodes.yaml" \ @@ -62,6 +75,7 @@ for _ in $(seq 1 30); do echo "" echo "Stop: bash scripts/stop.sh" echo "Tests: bash scripts/run_integration_tests.sh" + echo "State: volume '$STATE_VOLUME' (kept across stop/start)" exit 0 fi sleep 2 diff --git a/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/stop.sh b/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/stop.sh index fb3fa3de6..2ad4d0bda 100755 --- a/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/stop.sh +++ b/src/ros2_medkit_plugins/ros2_medkit_opcua/docker/scripts/stop.sh @@ -1,4 +1,14 @@ #!/usr/bin/env bash +# Stop the OpenPLC + medkit gateway demo. +# +# The gateway's state volume is deliberately LEFT IN PLACE. It holds the entity +# freeze frames, faults.db and the rosbags, and a freeze-frame is only worth +# anything if it outlives the process that took it. This script removes the +# container, so state kept in the container's writable layer would go with it. +STATE_VOLUME="${OPCUA_DEMO_STATE_VOLUME:-ros2-medkit-opcua-state}" + docker rm -f gateway openplc 2>/dev/null docker network rm plc-demo 2>/dev/null echo "Stopped." +echo "State kept in volume '$STATE_VOLUME'. Purge it with:" +echo " docker volume rm $STATE_VOLUME"