diff --git a/docker/postgres-compose.yaml b/docker/postgres-compose.yaml new file mode 100644 index 000000000..091c98d40 --- /dev/null +++ b/docker/postgres-compose.yaml @@ -0,0 +1,11 @@ +services: + db: + image: postgres:18 + restart: unless-stopped + container_name: ros2_medkit_postgres_test + network_mode: host + shm_size: 128mb + environment: + POSTGRES_USER: user + POSTGRES_PASSWORD: password + POSTGRES_DB: ros2_medkit_faults_database diff --git a/docker/postgres14-compose.yaml b/docker/postgres14-compose.yaml new file mode 100644 index 000000000..952478baa --- /dev/null +++ b/docker/postgres14-compose.yaml @@ -0,0 +1,11 @@ +services: + db: + image: postgres:14 + restart: unless-stopped + container_name: ros2_medkit_postgres_test + network_mode: host + shm_size: 128mb + environment: + POSTGRES_USER: user + POSTGRES_PASSWORD: password + POSTGRES_DB: ros2_medkit_faults_database diff --git a/docs/config/fault-manager.rst b/docs/config/fault-manager.rst index 400de56c0..67b27efd5 100644 --- a/docs/config/fault-manager.rst +++ b/docs/config/fault-manager.rst @@ -18,8 +18,13 @@ Storage fault_manager: ros__parameters: - storage_type: "sqlite" # Storage backend: "sqlite" or "memory" + storage_type: "sqlite" # Storage backend: "sqlite" or "memory" or "postgres" database_path: "/var/lib/ros2_medkit/faults.db" # Path for sqlite storage + database_url: "" # The url supports the following two patterns: + # Pattern 1: "postgresql://user:password@localhost:5432/ros2_medkit_faults_database" + # Pattern 2: "host=localhost port=5432 dbname=rosmedkit_faults_database user=user password=password" + # The recommended way to configure the connection is to use the PostgreSQL environment variables `PGUSER`, `PGPASSWORD`, `PGHOST`, `PGPORT` and `PGDATABASE` + # This ensures that the connection info cannot leak through ROS parameters .. list-table:: :header-rows: 1 @@ -34,6 +39,9 @@ Storage * - ``database_path`` - ``/var/lib/ros2_medkit/faults.db`` - File path for SQLite database. Directory must exist and be writable. + * - ``database_url`` + - ```` + - Connection URL for PostgreSQL database. The PostgreSQL server must be running with the appropriate user and password. The recommended way to configure the connection is to use the PostgreSQL environment variables `PGUSER`, `PGPASSWORD`, `PGHOST`, `PGPORT` and `PGDATABASE` Debounce Settings ~~~~~~~~~~~~~~~~~ @@ -581,7 +589,7 @@ by default: with it off there is no table, no file and no write cost. * - ``audit_log.database_path`` - ``""`` - Where the audit database lives. Empty puts it beside the fault database, - or in memory when the fault store is itself in memory or not SQLite. + or in memory when the fault store is itself in memory or unknown storage type. Correlation Configuration ------------------------- diff --git a/src/ros2_medkit_fault_manager/CMakeLists.txt b/src/ros2_medkit_fault_manager/CMakeLists.txt index 064cf44f1..99b8e447f 100644 --- a/src/ros2_medkit_fault_manager/CMakeLists.txt +++ b/src/ros2_medkit_fault_manager/CMakeLists.txt @@ -12,9 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. -cmake_minimum_required(VERSION 3.8) +cmake_minimum_required(VERSION 3.14) project(ros2_medkit_fault_manager) +option(POSTGRES_SUPPORT "Enable PostgreSQL support" OFF) + set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_EXPORT_COMPILE_COMMANDS ON) @@ -32,6 +34,56 @@ find_package(ament_cmake REQUIRED) find_package(rclcpp REQUIRED) find_package(ros2_medkit_msgs REQUIRED) find_package(ros2_medkit_serialization REQUIRED) +if(POSTGRES_SUPPORT) + find_package(PostgreSQL REQUIRED) + include(FetchContent) + + fetchcontent_declare(pqxx + GIT_REPOSITORY https://github.com/jtv/libpqxx.git + GIT_TAG 7.10.7 + GIT_SHALLOW TRUE + ) + + # Save medkit flags to restore them later. Not required per-se, but doesn't hurt to handle them either + set(_medkit_saved_cxx_flags "${CMAKE_CXX_FLAGS}") + get_directory_property(_medkit_saved_opts COMPILE_OPTIONS) + string(REGEX REPLACE "(^| )-W[^ ]*" " " CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS}") + set_directory_properties(PROPERTIES COMPILE_OPTIONS "") + + # The following lines are just to future proof the handling. + # Currently not required, and needs CMake3.24 and CMake3.3 respectively (without breaking older versions) + set(CMAKE_COMPILE_WARNING_AS_ERROR OFF) + unset(CMAKE_CXX_INCLUDE_WHAT_YOU_USE) + + # ros2_medkit_clang_tidy is called later, but guard pqxx either way + unset(CMAKE_CXX_CLANG_TIDY) + # pqxx appears to avoid building tests and docs when compiled externally as a dependency + # The lines below ensure that it doesn't + set(SKIP_BUILD_TEST ON) + set(BUILD_DOC OFF) + set(BUILD_TOOLS OFF) + + fetchcontent_makeavailable(pqxx) + + # Restore medkit flags now that pqxx is compiled + set(CMAKE_CXX_FLAGS "${_medkit_saved_cxx_flags}") + set_directory_properties(PROPERTIES COMPILE_OPTIONS "${_medkit_saved_opts}") + + # Make headers available to SYSTEM + # Once CMake3.25 is globally supported (Ubuntu 22.04 ships with CMake3.22) + # the snippet below could be replaced with fetchcontent_declare(pqxx ... SYSTEM) + foreach(_pqxx_tgt pqxx pqxx_shared pqxx_static) + if(TARGET ${_pqxx_tgt}) + target_compile_options(${_pqxx_tgt} PRIVATE -w) + get_target_property(_pqxx_inc ${_pqxx_tgt} INTERFACE_INCLUDE_DIRECTORIES) + if(_pqxx_inc) + set_target_properties(${_pqxx_tgt} PROPERTIES + INTERFACE_SYSTEM_INCLUDE_DIRECTORIES "${_pqxx_inc}") + endif() + endif() + endforeach() +endif() + find_package(SQLite3 REQUIRED) find_package(nlohmann_json REQUIRED) # OpenSSL EVP SHA-256 for the tamper-evident audit log hash chain @@ -45,19 +97,27 @@ find_package(rosbag2_storage REQUIRED) medkit_detect_compat_defs() # Library target (shared between executable and tests) +set(FAULT_MANAGER_FILES + src/capture_thread_pool.cpp + src/fault_manager_node.cpp + src/fault_storage.cpp + src/sqlite_fault_storage.cpp + src/fault_audit_log.cpp + src/snapshot_capture.cpp + src/rosbag_capture.cpp + src/correlation/types.cpp + src/correlation/config_parser.cpp + src/correlation/pattern_matcher.cpp + src/correlation/correlation_engine.cpp + src/entity_threshold_resolver.cpp +) +if(POSTGRES_SUPPORT) + list(APPEND FAULT_MANAGER_FILES + src/postgres_fault_storage.cpp + ) +endif() add_library(fault_manager_lib STATIC - src/capture_thread_pool.cpp - src/fault_manager_node.cpp - src/fault_storage.cpp - src/sqlite_fault_storage.cpp - src/fault_audit_log.cpp - src/snapshot_capture.cpp - src/rosbag_capture.cpp - src/correlation/types.cpp - src/correlation/config_parser.cpp - src/correlation/pattern_matcher.cpp - src/correlation/correlation_engine.cpp - src/entity_threshold_resolver.cpp + ${FAULT_MANAGER_FILES} ) target_include_directories(fault_manager_lib PUBLIC @@ -73,11 +133,20 @@ medkit_target_dependencies(fault_manager_lib PUBLIC rosbag2_storage ) +set(FAULT_MANAGER_TARGET_LIBS + SQLite::SQLite3 + nlohmann_json::nlohmann_json + yaml-cpp::yaml-cpp + OpenSSL::Crypto +) + +if(POSTGRES_SUPPORT) + list(APPEND FAULT_MANAGER_TARGET_LIBS + pqxx + ) +endif() target_link_libraries(fault_manager_lib PUBLIC - SQLite::SQLite3 - nlohmann_json::nlohmann_json - yaml-cpp::yaml-cpp - OpenSSL::Crypto + ${FAULT_MANAGER_TARGET_LIBS} ) medkit_apply_compat_defs(fault_manager_lib) @@ -136,6 +205,13 @@ if(BUILD_TESTING) target_link_libraries(test_sqlite_storage fault_manager_lib) medkit_target_dependencies(test_sqlite_storage rclcpp ros2_medkit_msgs) + # PostgreSQL storage tests +if(POSTGRES_SUPPORT) + medkit_add_gtest(test_postgres_storage test/test_postgres_storage.cpp) + target_link_libraries(test_postgres_storage fault_manager_lib) + medkit_target_dependencies(test_postgres_storage rclcpp ros2_medkit_msgs) +endif() + # Rosbag retention parity: every assertion runs against both storage backends. medkit_add_gtest(test_rosbag_storage_parity test/test_rosbag_storage_parity.cpp) target_link_libraries(test_rosbag_storage_parity fault_manager_lib) @@ -232,4 +308,9 @@ if(BUILD_TESTING) ros2_medkit_relax_vendor_warnings() endif() +if(POSTGRES_SUPPORT) + target_compile_definitions(fault_manager_lib PUBLIC POSTGRES_SUPPORT=1) +else() + target_compile_definitions(fault_manager_lib PUBLIC POSTGRES_SUPPORT=0) +endif() ament_package() diff --git a/src/ros2_medkit_fault_manager/README.md b/src/ros2_medkit_fault_manager/README.md index 977f5a8c4..edf787484 100644 --- a/src/ros2_medkit_fault_manager/README.md +++ b/src/ros2_medkit_fault_manager/README.md @@ -37,11 +37,11 @@ ros2 service call /fault_manager/clear_fault ros2_medkit_msgs/srv/ClearFault \ ## Services -| Service | Type | Description | -|---------|------|-------------| -| `~/report_fault` | `ros2_medkit_msgs/srv/ReportFault` | Report a fault occurrence | -| `~/list_faults` | `ros2_medkit_msgs/srv/ListFaults` | Query faults with filtering | -| `~/clear_fault` | `ros2_medkit_msgs/srv/ClearFault` | Clear/acknowledge a fault | +| Service | Type | Description | +| ----------------- | ----------------------------------- | ------------------------------- | +| `~/report_fault` | `ros2_medkit_msgs/srv/ReportFault` | Report a fault occurrence | +| `~/list_faults` | `ros2_medkit_msgs/srv/ListFaults` | Query faults with filtering | +| `~/clear_fault` | `ros2_medkit_msgs/srv/ClearFault` | Clear/acknowledge a fault | | `~/get_snapshots` | `ros2_medkit_msgs/srv/GetSnapshots` | Get topic snapshots for a fault | ## Features @@ -60,16 +60,17 @@ ros2 service call /fault_manager/clear_fault ros2_medkit_msgs/srv/ClearFault \ ## Parameters -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `storage_type` | string | `"sqlite"` | Storage backend: `"sqlite"` or `"memory"` | -| `database_path` | string | `"/var/lib/ros2_medkit/faults.db"` | Path to SQLite database file | -| `confirmation_threshold` | int | `-1` | Counter value at which faults are confirmed | -| `healing_enabled` | bool | `false` | Enable automatic healing via PASSED events | -| `healing_threshold` | int | `3` | Counter value at which faults are healed | -| `auto_confirm_after_sec` | double | `0.0` | Auto-confirm PREFAILED faults after timeout (0 = disabled) | -| `entity_thresholds.config_file` | string | `""` | Path to YAML file with per-entity debounce threshold overrides | -| `near_miss.max_per_fault` | int | `200` | Near-miss entries retained per fault code, oldest evicted first (0 = unlimited) | +| Parameter | Type | Default | Description | +| ------------------------------- | ------ | ---------------------------------- | ------------------------------------------------------------------------------- | +| `storage_type` | string | `"sqlite"` | Storage backend: `"sqlite"` or `"memory"` or `"postgres"` | +| `database_path` | string | `"/var/lib/ros2_medkit/faults.db"` | Path to SQLite database file | +| `database_url` | string | `""` | Connection URL to the PostgreSQL database | +| `confirmation_threshold` | int | `-1` | Counter value at which faults are confirmed | +| `healing_enabled` | bool | `false` | Enable automatic healing via PASSED events | +| `healing_threshold` | int | `3` | Counter value at which faults are healed | +| `auto_confirm_after_sec` | double | `0.0` | Auto-confirm PREFAILED faults after timeout (0 = disabled) | +| `entity_thresholds.config_file` | string | `""` | Path to YAML file with per-entity debounce threshold overrides | +| `near_miss.max_per_fault` | int | `200` | Near-miss entries retained per fault code, oldest evicted first (0 = unlimited) | ### Snapshot Parameters @@ -79,28 +80,30 @@ Each confirm also writes a **freeze-frame**: a single compact JSON object mappin Under a fault storm, captures are bounded by a worker pool (`capture_pool_size`) draining a bounded queue (`capture_queue_depth`); excess captures are dropped per `capture_queue_full_policy` and logged (throttled). The pool is shared and is created when snapshots **or** rosbag is enabled, so these parameters bound both. `capture_pool_size` parallelizes freeze-frame snapshot capture only - rosbag stays single-writer regardless of pool size, and correlated faults confirming inside one post-roll window share a single recording. -That single-writer property also shapes what each fault of a burst gets. Nothing is buffered while a post-fault window is open (messages go straight into the open bag), and the flush that opened that bag already emptied the ring buffer, so a fault confirming right *after* the window closes has no pre-fault history available. What a confirmation gets is decided by the buffer, so a fault arriving before any captured topic has published lands the same way. It gets a **post-fault-only bag**: its own recording holding just its `duration_after_sec` window, entered through the same post-roll state machine, so later faults of the burst attach to it normally. With `duration_after_sec: 0` there is no window to record into and such a fault gets no bag; if the bag cannot be written at all, no recording is opened and no metadata row is stored. The `duration_sec` on a stored bag is the span the recording was open rather than the configured windows, so a post-fault-only bag usually reports roughly `duration_after_sec` where a full one reports its buffered history too. It is a recording span, not a content span: a window during which nothing was published still reports the seconds it covered. It can also exceed `duration_sec + duration_after_sec`, because the ring buffer is pruned only when a message arrives - a topic that stops publishing keeps its last window buffered until the next confirmation flushes it, which is deliberate for a black box. See `docs/config/fault-manager.rst` for the full lifecycle. - -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `snapshots.enabled` | bool | `true` | Enable/disable snapshot capture | -| `snapshots.background_capture` | bool | `false` | Use background subscriptions (caches latest message) vs on-demand capture | -| `snapshots.timeout_sec` | double | `1.0` | Timeout waiting for topic message (on-demand mode) | -| `snapshots.max_message_size` | int | `65536` | Maximum message size in bytes (larger messages skipped) | -| `snapshots.default_topics` | string[] | `[]` | Topics to capture for all faults | -| `snapshots.config_file` | string | `""` | Path to YAML config for `fault_specific` and `patterns` | -| `snapshots.recapture_cooldown_sec` | double | `60.0` | Min seconds between captures for the same fault code. | -| `snapshots.max_per_fault` | int | `10` | Max snapshots retained per fault. | -| `snapshots.capture_pool_size` | int | `2` | Max concurrent capture threads under a fault storm (>= 1). Parallelizes snapshot capture only; rosbag stays single-writer. | -| `snapshots.capture_queue_depth` | int | `16` | Max pending captures before the full-queue policy applies (>= 1). | -| `snapshots.capture_queue_full_policy` | string | `reject_newest` | Policy when the queue is full: `reject_newest` or `drop_oldest`. | +That single-writer property also shapes what each fault of a burst gets. Nothing is buffered while a post-fault window is open (messages go straight into the open bag), and the flush that opened that bag already emptied the ring buffer, so a fault confirming right _after_ the window closes has no pre-fault history available. What a confirmation gets is decided by the buffer, so a fault arriving before any captured topic has published lands the same way. It gets a **post-fault-only bag**: its own recording holding just its `duration_after_sec` window, entered through the same post-roll state machine, so later faults of the burst attach to it normally. With `duration_after_sec: 0` there is no window to record into and such a fault gets no bag; if the bag cannot be written at all, no recording is opened and no metadata row is stored. The `duration_sec` on a stored bag is the span the recording was open rather than the configured windows, so a post-fault-only bag usually reports roughly `duration_after_sec` where a full one reports its buffered history too. It is a recording span, not a content span: a window during which nothing was published still reports the seconds it covered. It can also exceed `duration_sec + duration_after_sec`, because the ring buffer is pruned only when a message arrives - a topic that stops publishing keeps its last window buffered until the next confirmation flushes it, which is deliberate for a black box. See `docs/config/fault-manager.rst` for the full lifecycle. + +| Parameter | Type | Default | Description | +| ------------------------------------- | -------- | --------------- | -------------------------------------------------------------------------------------------------------------------------- | +| `snapshots.enabled` | bool | `true` | Enable/disable snapshot capture | +| `snapshots.background_capture` | bool | `false` | Use background subscriptions (caches latest message) vs on-demand capture | +| `snapshots.timeout_sec` | double | `1.0` | Timeout waiting for topic message (on-demand mode) | +| `snapshots.max_message_size` | int | `65536` | Maximum message size in bytes (larger messages skipped) | +| `snapshots.default_topics` | string[] | `[]` | Topics to capture for all faults | +| `snapshots.config_file` | string | `""` | Path to YAML config for `fault_specific` and `patterns` | +| `snapshots.recapture_cooldown_sec` | double | `60.0` | Min seconds between captures for the same fault code. | +| `snapshots.max_per_fault` | int | `10` | Max snapshots retained per fault. | +| `snapshots.capture_pool_size` | int | `2` | Max concurrent capture threads under a fault storm (>= 1). Parallelizes snapshot capture only; rosbag stays single-writer. | +| `snapshots.capture_queue_depth` | int | `16` | Max pending captures before the full-queue policy applies (>= 1). | +| `snapshots.capture_queue_full_policy` | string | `reject_newest` | Policy when the queue is full: `reject_newest` or `drop_oldest`. | **Topic Resolution Priority:** + 1. `fault_specific` - Exact match for fault code (configured via YAML config file) 2. `patterns` - Regex pattern match (configured via YAML config file) 3. `default_topics` - Fallback for all faults **Example YAML config file** (`snapshots.yaml`): + ```yaml fault_specific: MOTOR_OVERHEAT: @@ -135,6 +138,8 @@ format used by black-box capture (`snapshots.rosbag.format`, see Rosbag Capture **Memory**: Faults are stored in memory only. Useful for testing or when persistence is not required. +**PostgreSQL**: Faults are stored in an external PostgreSQL server and survive node restarts. The audit log stays a local SQLite file next to `database_path`. + ## Near-Miss Series A **near miss** is a FAILED report that moved the debounce counter without the fault ending up @@ -197,21 +202,21 @@ An optional append-only, hash-chained audit log records every fault state transi Each transition appends one immutable row holding `record_hash = sha256(prev_hash + canonical(event))` (OpenSSL EVP SHA-256), the `prev_hash` it links to, and a monotonic `seq`. The hash is computed once at insert and never recomputed. A persisted chain head lets the chain resume across restarts. The log is stored in its own SQLite database (separate from the fault store) and is treated as append-only: the manager only ever inserts rows, and `BEFORE UPDATE` / `BEFORE DELETE` triggers reject out-of-band edits (the guarded rotation prune excepted). -**Completeness is an integrity property.** `verify()` proves nothing was *deleted* from the chain, but it cannot prove a transition that was *never appended*. So a silently dropped append is a hole `verify()` can never see. Every transition on the write path is therefore audited (occurred, timer/threshold confirmations, auto-heal, and clears), and an append failure is never swallowed silently: it increments a dropped-writes counter and clears an "audit healthy" flag. **These are in-process signals only** (C++ getters on the node). This revision exposes no service/REST/health endpoint that surfaces audit health or lets an operator run `verify()` at runtime, so the signals are **not operator-observable at runtime yet** - a runtime read/verify/health surface is future work. With `audit_log.fail_closed` set, an append failure is re-raised as a **fail-FAST** error so a compliance-strict deployment learns the audit broke. This does **not** roll back the fault-state change that already committed: the audit log and the fault store are **separate SQLite databases**, so there is no cross-DB atomicity, and `fail_closed` is a broken-audit alarm requiring operator action, not a rollback. The default (`fail_closed=false`) keeps fault processing running; either way the in-process signals record the gap. +**Completeness is an integrity property.** `verify()` proves nothing was _deleted_ from the chain, but it cannot prove a transition that was _never appended_. So a silently dropped append is a hole `verify()` can never see. Every transition on the write path is therefore audited (occurred, timer/threshold confirmations, auto-heal, and clears), and an append failure is never swallowed silently: it increments a dropped-writes counter and clears an "audit healthy" flag. **These are in-process signals only** (C++ getters on the node). This revision exposes no service/REST/health endpoint that surfaces audit health or lets an operator run `verify()` at runtime, so the signals are **not operator-observable at runtime yet** - a runtime read/verify/health surface is future work. With `audit_log.fail_closed` set, an append failure is re-raised as a **fail-FAST** error so a compliance-strict deployment learns the audit broke. This does **not** roll back the fault-state change that already committed: the audit log and the fault store are **separate SQLite databases**, so there is no cross-DB atomicity, and `fail_closed` is a broken-audit alarm requiring operator action, not a rollback. The default (`fail_closed=false`) keeps fault processing running; either way the in-process signals record the gap. -`verify()` walks the persisted chain oldest-first and recomputes every link: editing a row breaks its `record_hash`, and deleting a row breaks the next row's `prev_hash` linkage. Deleting the newest row *while leaving the head untouched* is caught by the persisted-head check (the head is read straight from the DB). However, deleting the newest row(s) **and** repointing the head with a single `UPDATE audit_chain_head SET seq=..., record_hash=...` to the prior row's values costs no more than any other casual edit - it is **not** the "recompute the entire chain" the threat model below might suggest - and the truncated chain still verifies. There is no external record that a later `seq` ever existed, so this tail-truncation is undetectable by design. +`verify()` walks the persisted chain oldest-first and recomputes every link: editing a row breaks its `record_hash`, and deleting a row breaks the next row's `prev_hash` linkage. Deleting the newest row _while leaving the head untouched_ is caught by the persisted-head check (the head is read straight from the DB). However, deleting the newest row(s) **and** repointing the head with a single `UPDATE audit_chain_head SET seq=..., record_hash=...` to the prior row's values costs no more than any other casual edit - it is **not** the "recompute the entire chain" the threat model below might suggest - and the truncated chain still verifies. There is no external record that a later `seq` ever existed, so this tail-truncation is undetectable by design. -**Threat model (read this).** The chain is **unkeyed**, and the head and segment anchors live in the **same writable SQLite file** as the rows. `verify()` therefore catches edits or deletions that did **not** also recompute the chain - that is, casual or accidental tampering, and the bookkeeping bugs that would otherwise lose records. The append-only triggers are defense-in-depth: `audit_log` rejects out-of-band UPDATE/DELETE, `audit_anchors` carries the same guard-gated triggers so an out-of-band INSERT/UPDATE/DELETE of an anchor is rejected too, and the rotation-prune guard (`audit_prune_guard`) is itself protected by a trigger so an external writer cannot simply flip it open and then delete a prefix (or forge an anchor) - that flip is only permitted from the in-process connection that holds a per-connection temp marker. The single-row chain head (`audit_chain_head`) is intentionally **not** trigger-protected (a trigger there would block the legitimate head update inside the append transaction); a casual edit or delete of the head is instead caught by `verify()` via the seq/hash/head-mismatch checks. None of this stops an attacker with write access to the file: such an attacker can create the same temp marker or drop the triggers, and recompute the entire chain (head and anchors included) to forge a self-consistent history - and cheaper still, the tail-truncation above and the forged prefix-truncation below need no recompute at all. The triggers are **not** a security boundary - this is tamper-**evident**, not tamper-**proof**. True tamper-*proofing* requires a key or signature over the head (so it cannot be recomputed without the key) or external anchoring of the head hash to an append-only store you do not control; both are out of scope here and belong to the audit-log exporter / signing follow-up. +**Threat model (read this).** The chain is **unkeyed**, and the head and segment anchors live in the **same writable SQLite file** as the rows. `verify()` therefore catches edits or deletions that did **not** also recompute the chain - that is, casual or accidental tampering, and the bookkeeping bugs that would otherwise lose records. The append-only triggers are defense-in-depth: `audit_log` rejects out-of-band UPDATE/DELETE, `audit_anchors` carries the same guard-gated triggers so an out-of-band INSERT/UPDATE/DELETE of an anchor is rejected too, and the rotation-prune guard (`audit_prune_guard`) is itself protected by a trigger so an external writer cannot simply flip it open and then delete a prefix (or forge an anchor) - that flip is only permitted from the in-process connection that holds a per-connection temp marker. The single-row chain head (`audit_chain_head`) is intentionally **not** trigger-protected (a trigger there would block the legitimate head update inside the append transaction); a casual edit or delete of the head is instead caught by `verify()` via the seq/hash/head-mismatch checks. None of this stops an attacker with write access to the file: such an attacker can create the same temp marker or drop the triggers, and recompute the entire chain (head and anchors included) to forge a self-consistent history - and cheaper still, the tail-truncation above and the forged prefix-truncation below need no recompute at all. The triggers are **not** a security boundary - this is tamper-**evident**, not tamper-**proof**. True tamper-_proofing_ requires a key or signature over the head (so it cannot be recomputed without the key) or external anchoring of the head hash to an append-only store you do not control; both are out of scope here and belong to the audit-log exporter / signing follow-up. -**Retention/rotation**: when more than `audit_log.retention_max_records` rows are retained, the oldest segment is *sealed* (its final `seq` + hash are persisted as an anchor) and then pruned. The surviving tail still verifies because the oldest retained row links back to the sealed anchor. Only the anchor at the current prune boundary is kept - the same rotation drops older anchors - so `audit_anchors` stays bounded (one row) instead of growing one row per rotation. Because `verify()` treats any matching sealed anchor as a valid tail root, a **forged** prefix-truncation (an out-of-band actor deletes a prefix and inserts a matching anchor) is **indistinguishable** from legitimate pruning: "the surviving tail still verifies" therefore covers a forged truncation exactly as well as a real one. The guard-gated `audit_anchors` triggers raise the bar for this (casual/accidental only) but, like every trigger here, a write-capable adversary can drop them - so this stays tamper-**evident**, not tamper-**proof**. +**Retention/rotation**: when more than `audit_log.retention_max_records` rows are retained, the oldest segment is _sealed_ (its final `seq` + hash are persisted as an anchor) and then pruned. The surviving tail still verifies because the oldest retained row links back to the sealed anchor. Only the anchor at the current prune boundary is kept - the same rotation drops older anchors - so `audit_anchors` stays bounded (one row) instead of growing one row per rotation. Because `verify()` treats any matching sealed anchor as a valid tail root, a **forged** prefix-truncation (an out-of-band actor deletes a prefix and inserts a matching anchor) is **indistinguishable** from legitimate pruning: "the surviving tail still verifies" therefore covers a forged truncation exactly as well as a real one. The guard-gated `audit_anchors` triggers raise the bar for this (casual/accidental only) but, like every trigger here, a write-capable adversary can drop them - so this stays tamper-**evident**, not tamper-**proof**. -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `audit_log.enabled` | bool | `false` | Enable the tamper-evident audit log | -| `audit_log.transitions` | string | `"all"` | Which transitions to record: `"all"` (occurred/confirmed/healed/cleared) or `"confirmed_only"`. Lifecycle markers are always recorded. | -| `audit_log.database_path` | string | `""` | SQLite path. Empty => sibling `fault_audit.db` next to the fault DB (or `:memory:` for in-memory fault stores) | -| `audit_log.retention_max_records` | int | `0` | Seal + prune the oldest segment beyond this many retained records (0 = unlimited) | -| `audit_log.fail_closed` | bool | `false` | When `true`, an audit append failure is re-raised as a fail-FAST error signalling the audit chain is broken and needs operator action. It does **not** roll back the already-committed fault-state change (the fault store is a separate DB - no cross-DB atomicity). When `false`, the failure is logged and counted and fault processing continues. Either way the gap is recorded via the in-process dropped-writes / audit-healthy signals (not operator-observable at runtime yet). | +| Parameter | Type | Default | Description | +| --------------------------------- | ------ | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `audit_log.enabled` | bool | `false` | Enable the tamper-evident audit log | +| `audit_log.transitions` | string | `"all"` | Which transitions to record: `"all"` (occurred/confirmed/healed/cleared) or `"confirmed_only"`. Lifecycle markers are always recorded. | +| `audit_log.database_path` | string | `""` | SQLite path. Empty => sibling `fault_audit.db` next to the fault DB (or `:memory:` for in-memory fault stores) | +| `audit_log.retention_max_records` | int | `0` | Seal + prune the oldest segment beyond this many retained records (0 = unlimited) | +| `audit_log.fail_closed` | bool | `false` | When `true`, an audit append failure is re-raised as a fail-FAST error signalling the audit chain is broken and needs operator action. It does **not** roll back the already-committed fault-state change (the fault store is a separate DB - no cross-DB atomicity). When `false`, the failure is logged and counted and fault processing continues. Either way the gap is recorded via the in-process dropped-writes / audit-healthy signals (not operator-observable at runtime yet). | ## Usage @@ -321,12 +326,12 @@ PREFAILED -----> CONFIRMED -----> HEALED (retained) ### Status Reference -| Status | Description | -|--------|-------------| -| `PREFAILED` | Not yet confirmed. Usually a negative counter, but a fault that returns to 0 keeps the status it had | -| `CONFIRMED` | Fault is active and verified | -| `HEALED` | Resolved via PASSED events (if healing enabled) | -| `CLEARED` | Manually acknowledged via `~/clear_fault` | +| Status | Description | +| ----------- | ----------------------------------------------- | +| `PREFAILED` | Debounce counter < 0, not yet confirmed | +| `CONFIRMED` | Fault is active and verified | +| `HEALED` | Resolved via PASSED events (if healing enabled) | +| `CLEARED` | Manually acknowledged via `~/clear_fault` | ### Testing Debounce @@ -374,17 +379,17 @@ ros2 run ros2_medkit_fault_manager fault_manager_node --ros-args \ ### Correlation Parameters -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `correlation.config_file` | string | `""` | Path to correlation YAML config (empty = disabled) | -| `correlation.cleanup_interval_sec` | double | `5.0` | Interval for cleaning up expired pending correlations (seconds) | +| Parameter | Type | Default | Description | +| ---------------------------------- | ------ | ------- | --------------------------------------------------------------- | +| `correlation.config_file` | string | `""` | Path to correlation YAML config (empty = disabled) | +| `correlation.cleanup_interval_sec` | double | `5.0` | Interval for cleaning up expired pending correlations (seconds) | ### Configuration File Format ```yaml correlation: enabled: true - default_window_ms: 500 # Default time window for symptom detection + default_window_ms: 500 # Default time window for symptom detection # Reusable fault patterns (supports wildcards with *) patterns: @@ -405,8 +410,8 @@ correlation: symptoms: - pattern: motor_errors - pattern: drive_faults - window_ms: 1000 # Symptoms within 1s of root cause - mute_symptoms: true # Don't publish symptom events + window_ms: 1000 # Symptoms within 1s of root cause + mute_symptoms: true # Don't publish symptom events auto_clear_with_root: true # Clear symptoms when root cause clears # Auto-cluster rule: Group communication errors @@ -415,15 +420,16 @@ correlation: mode: auto_cluster match: - pattern: comm_errors - min_count: 3 # Need 3 faults to form cluster - window_ms: 500 # Within 500ms - show_as_single: true # Only show representative fault - representative: highest_severity # first | most_recent | highest_severity + min_count: 3 # Need 3 faults to form cluster + window_ms: 500 # Within 500ms + show_as_single: true # Only show representative fault + representative: highest_severity # first | most_recent | highest_severity ``` ### Pattern Wildcards Patterns support `*` wildcard matching: + - `MOTOR_*` matches `MOTOR_COMM`, `MOTOR_TIMEOUT`, `MOTOR_DRIVE_FAULT` - `*_COMM_*` matches `MOTOR_COMM_FL`, `SENSOR_COMM_TIMEOUT` - `*_TIMEOUT` matches `MOTOR_TIMEOUT`, `SENSOR_TIMEOUT` @@ -439,6 +445,7 @@ ros2 service call /fault_manager/list_faults ros2_medkit_msgs/srv/ListFaults \ ``` Response includes: + - `muted_count`: Number of muted symptom faults - `cluster_count`: Number of active fault clusters - `muted_faults[]`: Details of muted faults (when `include_muted=true`) @@ -447,10 +454,12 @@ Response includes: ### REST API (via Gateway) Query parameters for GET `/api/v1/faults`: + - `include_muted=true`: Include muted fault details in response - `include_clusters=true`: Include cluster details in response Response fields: + ```json { "faults": [...], @@ -482,6 +491,7 @@ Response fields: ``` When clearing a root cause fault, `auto_cleared_codes` lists symptoms that were auto-cleared: + ```json { "status": "success", diff --git a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_manager_node.hpp b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_manager_node.hpp index 9157102d9..5a4806163 100644 --- a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_manager_node.hpp +++ b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_manager_node.hpp @@ -205,6 +205,7 @@ class FaultManagerNode : public rclcpp::Node { std::string storage_type_; std::string database_path_; + std::string database_url_; int32_t confirmation_threshold_{-1}; bool healing_enabled_{false}; int32_t healing_threshold_{3}; diff --git a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_storage.hpp b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_storage.hpp index 3bc60dc35..7f0d412e3 100644 --- a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_storage.hpp +++ b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/fault_storage.hpp @@ -198,6 +198,23 @@ struct RosbagFileInfo { /// Abstract interface for fault storage backends class FaultStorage { public: + // A custom exception to be used by fault storage implementations to signify an connection exception that should be + // ignored + class IgnorableConnectionException : public std::exception { + protected: + std::string message; + + public: + explicit IgnorableConnectionException(const std::string & msg = "IgnorableConnectionException") : message(msg) { + } + + const char * what() const noexcept override { + return message.c_str(); + } + + virtual ~IgnorableConnectionException() = default; + }; + virtual ~FaultStorage() = default; /// Set debounce configuration @@ -226,12 +243,12 @@ class FaultStorage { /// @param statuses List of statuses to include (empty = CONFIRMED only) /// @return Vector of matching faults virtual std::vector list_faults(bool filter_by_severity, uint8_t severity, - const std::vector & statuses) const = 0; + const std::vector & statuses) = 0; /// Get a single fault by fault_code /// @param fault_code The fault code to look up /// @return The fault if found, nullopt otherwise - virtual std::optional get_fault(const std::string & fault_code) const = 0; + virtual std::optional get_fault(const std::string & fault_code) = 0; /// Clear a fault by fault_code (manual acknowledgment). Drops the fault's per-topic snapshots; /// the freeze-frame and the near-miss series are RETAINED, because they outlive a single fault @@ -241,10 +258,10 @@ class FaultStorage { virtual bool clear_fault(const std::string & fault_code) = 0; /// Get total number of stored faults - virtual size_t size() const = 0; + virtual size_t size() = 0; /// Check if a fault exists - virtual bool contains(const std::string & fault_code) const = 0; + virtual bool contains(const std::string & fault_code) = 0; /// Check and confirm PREFAILED faults that have been pending too long (time-based confirmation) /// @param current_time Current timestamp for age calculation @@ -312,7 +329,7 @@ class FaultStorage { /// @param topic_filter Optional topic filter (empty = all topics) /// @return Vector of snapshots for the fault virtual std::vector get_snapshots(const std::string & fault_code, - const std::string & topic_filter = "") const = 0; + const std::string & topic_filter = "") = 0; /// Highest capture_id any stored snapshot holds, across every fault (0 when none). /// @@ -320,7 +337,7 @@ class FaultStorage { /// hand out ids BELOW the ones already on disk: the eviction that protects /// MAX(capture_id) would then guard an old set and drop the one just written. /// SnapshotCapture seeds its counter from this at construction. - virtual int64_t get_max_capture_id() const { + virtual int64_t get_max_capture_id() { return 0; } @@ -337,7 +354,7 @@ class FaultStorage { /// @param fault_code The fault code to look up /// @return The freeze-frame if one was captured, nullopt otherwise (including fault /// codes with no capture configured, which never get a row) - virtual std::optional get_freeze_frame(const std::string & fault_code) const = 0; + virtual std::optional get_freeze_frame(const std::string & fault_code) = 0; /// Set the maximum number of near-miss entries retained per fault code. /// @@ -356,7 +373,7 @@ class FaultStorage { /// The series survives clear_fault; an unknown or never-near-missed code returns empty. /// @param fault_code The fault code to look up /// @return The retained near-miss entries in chronological order - virtual std::vector get_near_misses(const std::string & fault_code) const = 0; + virtual std::vector get_near_misses(const std::string & fault_code) = 0; /// Store rosbag file metadata for a fault /// @param info The rosbag file info to store (replaces any existing entry for fault_code) @@ -389,15 +406,15 @@ class FaultStorage { /// serves an arbitrary recording, which no test catches reliably. /// @param fault_code The fault code to get rosbag for /// @return Rosbag file info if exists, nullopt otherwise - virtual std::optional get_rosbag_file(const std::string & fault_code) const = 0; + virtual std::optional get_rosbag_file(const std::string & fault_code) = 0; /// Every recording of a fault, newest first. - virtual std::vector get_rosbag_files(const std::string & fault_code) const = 0; + virtual std::vector get_rosbag_files(const std::string & fault_code) = 0; /// Every row of one recording - one per fault the recording covers. Backs the /// bulk-data download and the entity authorization scope check, both of which /// start from a recording id and need the faults behind it. - virtual std::vector get_rosbag_files_by_recording(const std::string & recording_id) const = 0; + virtual std::vector get_rosbag_files_by_recording(const std::string & recording_id) = 0; /// Delete one whole recording: every fault's link to it, and the bag. This is the /// right unit for quota eviction and for a bag that has vanished from disk - both @@ -431,20 +448,20 @@ class FaultStorage { /// Get total size of all stored rosbag files in bytes, counting a shared /// recording once regardless of how many faults reference it /// @return Total size in bytes - virtual size_t get_total_rosbag_storage_bytes() const = 0; + virtual size_t get_total_rosbag_storage_bytes() = 0; /// Get all rosbag files ordered by creation time (oldest first) /// @return Vector of rosbag file info - virtual std::vector get_all_rosbag_files() const = 0; + virtual std::vector get_all_rosbag_files() = 0; /// Get rosbags for all faults associated with an entity /// @param entity_fqn The entity's fully qualified name to filter by /// @return Vector of rosbag file info for faults reported by this entity - virtual std::vector list_rosbags_for_entity(const std::string & entity_fqn) const = 0; + virtual std::vector list_rosbags_for_entity(const std::string & entity_fqn) = 0; /// Get all stored faults regardless of status (for filtering) /// @return Vector of all faults in storage - virtual std::vector get_all_faults() const = 0; + virtual std::vector get_all_faults() = 0; /// One-time startup cleanup: reclassify HEALED faults as CLEARED. Called when healing is disabled, /// so a HEALED row left by a previous (healing-enabled) run does not behave inconsistently under @@ -475,15 +492,15 @@ class InMemoryFaultStorage : public FaultStorage { const rclcpp::Time & timestamp, const DebounceConfig & config) override; std::vector list_faults(bool filter_by_severity, uint8_t severity, - const std::vector & statuses) const override; + const std::vector & statuses) override; - std::optional get_fault(const std::string & fault_code) const override; + std::optional get_fault(const std::string & fault_code) override; bool clear_fault(const std::string & fault_code) override; - size_t size() const override; + size_t size() override; - bool contains(const std::string & fault_code) const override; + bool contains(const std::string & fault_code) override; std::vector check_time_based_confirmation(const rclcpp::Time & current_time) override; @@ -494,30 +511,30 @@ class InMemoryFaultStorage : public FaultStorage { void store_snapshot(const SnapshotData & snapshot) override; void store_snapshots(const std::vector & snapshots) override; std::vector get_snapshots(const std::string & fault_code, - const std::string & topic_filter = "") const override; - int64_t get_max_capture_id() const override; + const std::string & topic_filter = "") override; + int64_t get_max_capture_id() override; void store_freeze_frame(const FreezeFrameData & frame) override; - std::optional get_freeze_frame(const std::string & fault_code) const override; + std::optional get_freeze_frame(const std::string & fault_code) override; void set_max_rosbags_per_fault(size_t max_count) override; size_t set_max_near_misses_per_fault(size_t max_count) override; - std::vector get_near_misses(const std::string & fault_code) const override; + std::vector get_near_misses(const std::string & fault_code) override; void store_rosbag_file(const RosbagFileInfo & info) override; /// All-or-nothing, as the base class requires: the batch is built beside the live /// map and swapped in, so a throw leaves the store exactly as it was. void store_rosbag_files(const std::vector & infos) override; - std::optional get_rosbag_file(const std::string & fault_code) const override; - std::vector get_rosbag_files(const std::string & fault_code) const override; - std::vector get_rosbag_files_by_recording(const std::string & recording_id) const override; + std::optional get_rosbag_file(const std::string & fault_code) override; + std::vector get_rosbag_files(const std::string & fault_code) override; + std::vector get_rosbag_files_by_recording(const std::string & recording_id) override; bool delete_rosbag_file(const std::string & fault_code) override; size_t delete_rosbag_recording(const std::string & recording_id) override; - size_t get_total_rosbag_storage_bytes() const override; - std::vector get_all_rosbag_files() const override; - std::vector list_rosbags_for_entity(const std::string & entity_fqn) const override; - std::vector get_all_faults() const override; + size_t get_total_rosbag_storage_bytes() override; + std::vector get_all_rosbag_files() override; + std::vector list_rosbags_for_entity(const std::string & entity_fqn) override; + std::vector get_all_faults() override; std::vector reclassify_healed_as_cleared() override; private: @@ -534,7 +551,7 @@ class InMemoryFaultStorage : public FaultStorage { /// only be unlinked once the last of them is gone. Caller holds mutex_. /// Whether any row at all still references @p file_path. Caller holds mutex_. - bool path_referenced(const std::string & file_path) const; + bool path_referenced(const std::string & file_path); mutable std::mutex mutex_; std::map faults_; diff --git a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/postgres_fault_storage.hpp b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/postgres_fault_storage.hpp new file mode 100644 index 000000000..033401567 --- /dev/null +++ b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/postgres_fault_storage.hpp @@ -0,0 +1,190 @@ +// Copyright 2026 gstavrinos +// +// 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 "ros2_medkit_fault_manager/fault_storage.hpp" + +namespace ros2_medkit_fault_manager { + +/// PostgreSQL-based fault storage implementation with persistence +/// Thread-safe implementation using mutex protection on connection access +class PgFaultStorage : public FaultStorage { + public: + using FaultStorage::IgnorableConnectionException; + /// Create PostgreSQL fault storage + /// @param conn_info Connection string or DSN (e.g., + /// "postgresql://user:password@localhost:5432/ros2_medkit_faults_database") + /// @throws std::runtime_error if database cannot be connected to or initialized + explicit PgFaultStorage(const std::string & conn_info); + + /// Create PostgreSQL fault storage + /// @param conn_info Connection string or DSN (e.g., + /// "postgresql://user:password@localhost:5432/ros2_medkit_faults_database") + /// @param max_retries Maximum retries when attempting to reconnect + /// @param reconnection_delay_ms The delay in milliseconds between each recconection attempt + /// @throws std::runtime_error if database cannot be connected to or initialized + explicit PgFaultStorage(const std::string & conn_info, const int max_retries, const unsigned reconnection_delay_ms); + + /// Destructor - closes database connection + ~PgFaultStorage() override; + + // Non-copyable, non-movable (owns PostgreSQL connection) + PgFaultStorage(const PgFaultStorage &) = delete; + PgFaultStorage & operator=(const PgFaultStorage &) = delete; + PgFaultStorage(PgFaultStorage &&) = delete; + PgFaultStorage & operator=(PgFaultStorage &&) = delete; + + void set_debounce_config(const DebounceConfig & config) override; + DebounceConfig get_debounce_config() const override; + + bool report_fault_event(const std::string & fault_code, uint8_t event_type, uint8_t severity, + const std::string & description, const std::string & source_id, + const rclcpp::Time & timestamp, const DebounceConfig & config) override; + + std::vector list_faults(bool filter_by_severity, uint8_t severity, + const std::vector & statuses) override; + + std::optional get_fault(const std::string & fault_code) override; + + bool clear_fault(const std::string & fault_code) override; + + size_t size() override; + + bool contains(const std::string & fault_code) override; + + std::vector check_time_based_confirmation(const rclcpp::Time & current_time) override; + + void set_max_snapshots_per_fault(size_t max_count) override; + void set_retain_snapshots_on_clear(bool retain) override; + bool retains_snapshots_on_clear() const override; + + void set_max_rosbags_per_fault(size_t max_count) override; + + void store_snapshot(const SnapshotData & snapshot) override; + void store_snapshots(const std::vector & snapshots) override; + std::vector get_snapshots(const std::string & fault_code, + const std::string & topic_filter = "") override; + int64_t get_max_capture_id() override; + + void store_freeze_frame(const FreezeFrameData & frame) override; + std::optional get_freeze_frame(const std::string & fault_code) override; + size_t set_max_near_misses_per_fault(size_t max_count) override; + std::vector get_near_misses(const std::string & fault_code) override; + + void store_rosbag_file(const RosbagFileInfo & info) override; + void store_rosbag_files(const std::vector & infos) override; + std::optional get_rosbag_file(const std::string & fault_code) override; + std::vector get_rosbag_files(const std::string & fault_code) override; + std::vector get_rosbag_files_by_recording(const std::string & recording_id) override; + bool delete_rosbag_file(const std::string & fault_code) override; + size_t delete_rosbag_recording(const std::string & recording_id) override; + size_t delete_rosbag_files(const std::vector & fault_codes) override; + size_t get_total_rosbag_storage_bytes() override; + std::vector get_all_rosbag_files() override; + std::vector list_rosbags_for_entity(const std::string & entity_fqn) override; + std::vector get_all_faults() override; + std::vector reclassify_healed_as_cleared() override; + + /// Get the connection info string used to initialize the database + const std::string & conn_info() const { + return conn_info_; + } + + /// Get the hostname of the PostgreSQL server + const std::string hostname() const { + return db_conn_->hostname(); + } + + /// Get the port of the PostgreSQL server + const std::string port() const { + return db_conn_->port(); + } + + /// Get the database of the PostgreSQL server + const std::string dbname() const { + return db_conn_->dbname(); + } + + private: + /// Wrapper the pqxx exec function + template + pqxx::result execute(pqxx::work & tx, const std::string & query, Args &&... args); + + /// The function that establishes connection + void ensure_connection(); + + /// Wrapper of the execute function to encapsulate the recconection mechanism + /// while keeping the transaction pqxx::work object safe + template + auto run_in_transaction(const char * what, Fn && fn) -> decltype(fn(std::declval())); + + /// Initialize database schema (create tables if they don't exist) + void initialize_schema(); + + /// Whether any fault at all still references @p file_path. Caller holds mutex_. + bool path_referenced(const std::string & file_path); + + /// store_rosbag_file body without taking mutex_. Caller holds mutex_ and + /// manages transaction scope. Returns replaced bag path if applicable. + std::vector store_rosbag_file_locked(const RosbagFileInfo & info, pqxx::work & tx); + + /// report_fault_event body without taking mutex_ or opening a transaction. Caller holds mutex_ + /// and supplies the transaction, so the fault row and any near-miss row commit together. + bool report_fault_event_locked(const std::string & fault_code, uint8_t event_type, uint8_t severity, + const std::string & description, const std::string & source_id, + const rclcpp::Time & timestamp, const DebounceConfig & config, pqxx::work & tx); + + /// Append one entry to the near-miss series and evict the oldest entries beyond + /// max_near_misses_per_fault_. Caller holds mutex_ and has already written the fault row. + /// @param fault_code The fault code that nearly confirmed + /// @param occurred_at_ns Timestamp of the report + /// @param debounce_counter Counter value after the report + /// @param config Debounce config the report was evaluated against + /// @param severity Severity carried by the report + /// @param source_id Reporting source + /// @param resulting_status Fault status after the report was applied + /// @param tx Transaction the fault row was written in + void record_near_miss_locked(const std::string & fault_code, int64_t occurred_at_ns, int32_t debounce_counter, + const DebounceConfig & config, uint8_t severity, const std::string & source_id, + const std::string & resulting_status, pqxx::work & tx); + + /// Deserialize JSON array string from PostgreSQL TEXT/JSONB field + static std::vector parse_json_array(const std::string & json_str); + + /// Serialize vector of strings to JSON array string (for PostgreSQL JSONB) + static std::string serialize_json_array(const std::vector & vec); + + std::string conn_info_; + std::unique_ptr db_conn_; + mutable std::mutex mutex_; + DebounceConfig config_; + size_t max_snapshots_per_fault_{0}; ///< 0 = unlimited + size_t max_near_misses_per_fault_{0}; ///< 0 = unlimited + bool retain_snapshots_on_clear_{false}; + /// Defaults to 1, the pre-#620 behaviour: a new recording replaces the old one. + /// 0 = unlimited, bounded only by max_total_storage_mb. + size_t max_rosbags_per_fault_{1}; + int max_retries_{1}; /// < 0 = unlimited + unsigned reconnection_delay_{500}; +}; + +} // namespace ros2_medkit_fault_manager diff --git a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/sqlite_fault_storage.hpp b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/sqlite_fault_storage.hpp index 3565febf4..4c0b5c52e 100644 --- a/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/sqlite_fault_storage.hpp +++ b/src/ros2_medkit_fault_manager/include/ros2_medkit_fault_manager/sqlite_fault_storage.hpp @@ -51,15 +51,15 @@ class SqliteFaultStorage : public FaultStorage { const rclcpp::Time & timestamp, const DebounceConfig & config) override; std::vector list_faults(bool filter_by_severity, uint8_t severity, - const std::vector & statuses) const override; + const std::vector & statuses) override; - std::optional get_fault(const std::string & fault_code) const override; + std::optional get_fault(const std::string & fault_code) override; bool clear_fault(const std::string & fault_code) override; - size_t size() const override; + size_t size() override; - bool contains(const std::string & fault_code) const override; + bool contains(const std::string & fault_code) override; std::vector check_time_based_confirmation(const rclcpp::Time & current_time) override; @@ -72,27 +72,27 @@ class SqliteFaultStorage : public FaultStorage { void store_snapshot(const SnapshotData & snapshot) override; void store_snapshots(const std::vector & snapshots) override; std::vector get_snapshots(const std::string & fault_code, - const std::string & topic_filter = "") const override; - int64_t get_max_capture_id() const override; + const std::string & topic_filter = "") override; + int64_t get_max_capture_id() override; void store_freeze_frame(const FreezeFrameData & frame) override; - std::optional get_freeze_frame(const std::string & fault_code) const override; + std::optional get_freeze_frame(const std::string & fault_code) override; size_t set_max_near_misses_per_fault(size_t max_count) override; - std::vector get_near_misses(const std::string & fault_code) const override; + std::vector get_near_misses(const std::string & fault_code) override; void store_rosbag_file(const RosbagFileInfo & info) override; void store_rosbag_files(const std::vector & infos) override; - std::optional get_rosbag_file(const std::string & fault_code) const override; - std::vector get_rosbag_files(const std::string & fault_code) const override; - std::vector get_rosbag_files_by_recording(const std::string & recording_id) const override; + std::optional get_rosbag_file(const std::string & fault_code) override; + std::vector get_rosbag_files(const std::string & fault_code) override; + std::vector get_rosbag_files_by_recording(const std::string & recording_id) override; bool delete_rosbag_file(const std::string & fault_code) override; size_t delete_rosbag_recording(const std::string & recording_id) override; size_t delete_rosbag_files(const std::vector & fault_codes) override; - size_t get_total_rosbag_storage_bytes() const override; - std::vector get_all_rosbag_files() const override; - std::vector list_rosbags_for_entity(const std::string & entity_fqn) const override; - std::vector get_all_faults() const override; + size_t get_total_rosbag_storage_bytes() override; + std::vector get_all_rosbag_files() override; + std::vector list_rosbags_for_entity(const std::string & entity_fqn) override; + std::vector get_all_faults() override; std::vector reclassify_healed_as_cleared() override; /// Get the database path @@ -125,7 +125,7 @@ class SqliteFaultStorage : public FaultStorage { /// only be unlinked once the last of them is gone. Caller holds mutex_. /// Whether any fault at all still references @p file_path. Caller holds mutex_. - bool path_referenced(const std::string & file_path) const; + bool path_referenced(const std::string & file_path); /// store_rosbag_file body without taking mutex_. Caller holds mutex_ and /// unlinks the returned replaced-bag path once the row change is durable. diff --git a/src/ros2_medkit_fault_manager/package.xml b/src/ros2_medkit_fault_manager/package.xml index 445d17ec9..a9c091904 100644 --- a/src/ros2_medkit_fault_manager/package.xml +++ b/src/ros2_medkit_fault_manager/package.xml @@ -15,6 +15,7 @@ ros2_medkit_msgs ros2_medkit_serialization libsqlite3-dev + libpqxx-dev nlohmann-json-dev libssl-dev rosbag2_cpp diff --git a/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp b/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp index 146f0e3b4..601ce4027 100644 --- a/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp +++ b/src/ros2_medkit_fault_manager/src/fault_manager_node.cpp @@ -27,6 +27,9 @@ #include #include "ros2_medkit_fault_manager/correlation/config_parser.hpp" +#ifdef POSTGRES_SUPPORT +#include "ros2_medkit_fault_manager/postgres_fault_storage.hpp" +#endif #include "ros2_medkit_fault_manager/sqlite_fault_storage.hpp" #include "ros2_medkit_fault_manager/time_utils.hpp" #include "ros2_medkit_msgs/msg/cluster_info.hpp" @@ -119,6 +122,7 @@ FaultManagerNode::FaultManagerNode(const rclcpp::NodeOptions & options) : Node(" // Declare and get parameters storage_type_ = declare_parameter("storage_type", "sqlite"); database_path_ = declare_parameter("database_path", "/var/lib/ros2_medkit/faults.db"); + database_url_ = declare_parameter("database_url", ""); auto confirmation_threshold_param = declare_parameter("confirmation_threshold", -1); if (confirmation_threshold_param > 0) { @@ -242,7 +246,13 @@ FaultManagerNode::FaultManagerNode(const rclcpp::NodeOptions & options) : Node(" // Apply near-miss retention bound to storage (0 = unlimited). Applying it also trims a series // that a previous run left over the bound, which deletes history for good, so say when it does. - const size_t evicted_near_misses = storage_->set_max_near_misses_per_fault(static_cast(max_near_misses)); + size_t evicted_near_misses = 0; + try { + evicted_near_misses = storage_->set_max_near_misses_per_fault(static_cast(max_near_misses)); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (evicted_near_misses > 0) { RCLCPP_WARN(get_logger(), "near_miss.max_per_fault=%ld dropped %zu stored near-miss entries that exceeded the bound. " @@ -271,12 +281,24 @@ FaultManagerNode::FaultManagerNode(const rclcpp::NodeOptions & options) : Node(" // fault is audited (the audit log is already constructed above); without this the reclassification // would be invisible to the audit log's verify(). if (!global_config_.healing_enabled) { - const auto reclassified = storage_->reclassify_healed_as_cleared(); + std::vector reclassified = {}; + try { + reclassified = storage_->reclassify_healed_as_cleared(); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (!reclassified.empty()) { if (audit_log_) { const int64_t reclassified_at_ns = get_wall_clock_time().nanoseconds(); for (const auto & fault_code : reclassified) { - auto fault = storage_->get_fault(fault_code); + std::optional fault; + try { + fault = storage_->get_fault(fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (fault) { audit_transition(kTransitionCleared, *fault, "startup_reclassify", reclassified_at_ns); } @@ -388,7 +410,12 @@ FaultManagerNode::FaultManagerNode(const rclcpp::NodeOptions & options) : Node(" // The cap lives in the storage backend, not in RosbagCapture: it has to be // atomic with the insert, the unlink has to happen after the backend's commit, // and the "is this bag still referenced" rule is backend-private. - storage_->set_max_rosbags_per_fault(snapshot_config.rosbag.max_bags_per_fault); + try { + storage_->set_max_rosbags_per_fault(snapshot_config.rosbag.max_bags_per_fault); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } rosbag_capture_ = std::make_shared(this, storage_.get(), snapshot_config.rosbag, snapshot_config); // The recapture cooldown gates the capture job as a whole, bags included, so it @@ -447,7 +474,13 @@ FaultManagerNode::FaultManagerNode(const rclcpp::NodeOptions & options) : Node(" // Create auto-confirmation timer if enabled if (auto_confirm_after_sec_ > 0.0) { auto_confirm_timer_ = create_wall_timer(std::chrono::seconds(1), [this]() { - const auto confirmed = storage_->check_time_based_confirmation(get_wall_clock_time()); + std::vector confirmed = {}; + try { + confirmed = storage_->check_time_based_confirmation(get_wall_clock_time()); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (confirmed.empty()) { return; } @@ -455,7 +488,13 @@ FaultManagerNode::FaultManagerNode(const rclcpp::NodeOptions & options) : Node(" // confirmations are invisible to the audit log's verify(). const int64_t confirmed_at_ns = get_wall_clock_time().nanoseconds(); for (const auto & fault_code : confirmed) { - auto fault = storage_->get_fault(fault_code); + std::optional fault; + try { + fault = storage_->get_fault(fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (fault) { audit_transition(kTransitionConfirmed, *fault, "auto_confirm_timer", confirmed_at_ns); // A timer-driven confirmation is a confirmation: it has to reach the @@ -549,6 +588,16 @@ std::unique_ptr FaultManagerNode::create_storage() { return std::make_unique(database_path_); } +#ifdef POSTGRES_SUPPORT + if (storage_type_ == "postgres") { + auto postgres_fault_storage = std::make_unique(database_url_); + RCLCPP_INFO(get_logger(), "Using PostgreSQL fault storage - Host: %s, Port: %s, Database: %s", + postgres_fault_storage->hostname().c_str(), postgres_fault_storage->port().c_str(), + postgres_fault_storage->dbname().c_str()); + return postgres_fault_storage; + } +#endif + RCLCPP_ERROR(get_logger(), "Unknown storage_type '%s', falling back to in-memory", storage_type_.c_str()); return std::make_unique(); } @@ -588,7 +637,7 @@ std::unique_ptr FaultManagerNode::create_audit_log() { } if (audit_path.empty()) { - if (database_path_ == ":memory:" || storage_type_ != "sqlite") { + if (database_path_ == ":memory:" || (storage_type_ != "sqlite" && storage_type_ != "postgres")) { audit_path = ":memory:"; } else { std::filesystem::path base(database_path_); @@ -804,7 +853,13 @@ void FaultManagerNode::handle_report_fault( } // Get status before update (if fault exists) - auto fault_before = storage_->get_fault(request->fault_code); + std::optional fault_before; + try { + fault_before = storage_->get_fault(request->fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } std::string status_before = fault_before ? fault_before->status : ""; // Resolve per-entity debounce config (longest-prefix match on source_id) @@ -813,13 +868,25 @@ void FaultManagerNode::handle_report_fault( // Report the fault event (use wall clock time, not sim time, for proper timestamps) const rclcpp::Time event_time = get_wall_clock_time(); - bool is_new = storage_->report_fault_event(request->fault_code, request->event_type, request->severity, - request->description, request->source_id, event_time, resolved_config); + bool is_new = false; + try { + is_new = storage_->report_fault_event(request->fault_code, request->event_type, request->severity, + request->description, request->source_id, event_time, resolved_config); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } response->accepted = true; // Get updated fault state to publish event - auto fault_after = storage_->get_fault(request->fault_code); + std::optional fault_after; + try { + fault_after = storage_->get_fault(request->fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (fault_after) { // Process through correlation engine (if enabled) // Only process FAILED events with correlation @@ -929,7 +996,12 @@ void FaultManagerNode::handle_report_fault( void FaultManagerNode::handle_list_faults( const std::shared_ptr & request, const std::shared_ptr & response) { - response->faults = storage_->list_faults(request->filter_by_severity, request->severity, request->statuses); + try { + response->faults = storage_->list_faults(request->filter_by_severity, request->severity, request->statuses); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } // Include correlation data if engine is enabled if (correlation_engine_) { @@ -1016,7 +1088,13 @@ void FaultManagerNode::handle_clear_fault( auto_cleared_codes = clear_result.auto_cleared_codes; } - bool cleared = storage_->clear_fault(request->fault_code); + bool cleared = false; + try { + cleared = storage_->clear_fault(request->fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } response->success = cleared; if (cleared) { @@ -1031,9 +1109,20 @@ void FaultManagerNode::handle_clear_fault( // Auto-clear correlated symptoms for (const auto & symptom_code : auto_cleared_codes) { - storage_->clear_fault(symptom_code); + try { + storage_->clear_fault(symptom_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (audit_log_) { - auto symptom = storage_->get_fault(symptom_code); + std::optional symptom; + try { + symptom = storage_->get_fault(symptom_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (symptom) { audit_transition(kTransitionCleared, *symptom, "clear_service", get_wall_clock_time().nanoseconds()); } @@ -1062,7 +1151,13 @@ void FaultManagerNode::handle_clear_fault( } // Publish EVENT_CLEARED - get the cleared fault to include in event - auto fault = storage_->get_fault(request->fault_code); + std::optional fault; + try { + fault = storage_->get_fault(request->fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (fault) { publish_fault_event(ros2_medkit_msgs::msg::FaultEvent::EVENT_CLEARED, *fault, auto_cleared_codes); audit_transition(kTransitionCleared, *fault, "clear_service", get_wall_clock_time().nanoseconds()); @@ -1098,7 +1193,13 @@ void FaultManagerNode::handle_get_fault(const std::shared_ptrget_fault(request->fault_code); + std::optional fault; + try { + fault = storage_->get_fault(request->fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } if (!fault) { response->success = false; response->error_message = "Fault not found: " + request->fault_code; @@ -1115,7 +1216,13 @@ void FaultManagerNode::handle_get_fault(const std::shared_ptrenvironment_data.extended_data_records = extended_records; // Get freeze frame snapshots from storage - auto stored_snapshots = storage_->get_snapshots(request->fault_code); + std::vector stored_snapshots = {}; + try { + stored_snapshots = storage_->get_snapshots(request->fault_code); + } catch (const FaultStorage::IgnorableConnectionException & e) { + RCLCPP_WARN(get_logger(), "Failed to connect with the fault storage server: %s", e.what()); + return; + } for (const auto & stored_snapshot : stored_snapshots) { ros2_medkit_msgs::msg::Snapshot snapshot; snapshot.type = ros2_medkit_msgs::msg::Snapshot::TYPE_FREEZE_FRAME; @@ -1136,7 +1243,13 @@ void FaultManagerNode::handle_get_fault(const std::shared_ptr