Nest x-medkit vendor fields on SOVD endpoint payloads

.github/workflows/quality.yml

@@ -217,7 +217,11 @@ jobs:
     runs-on: ubuntu-latest
     container:
       image: ubuntu:noble
-    timeout-minutes: 45
+    # Cold-cache ASan builds are ~33 min; with the 12-min test budget this
+    # left ~0 headroom and was already cancelling at thermal_protection on
+    # PR branches that miss the ccache restore key. 60 min covers a full
+    # cold-cache cycle so the cache always gets saved on the post step.
+    timeout-minutes: 60
     defaults:
       run:
         shell: bash

docs/api/rest.rst

@@ -1154,25 +1154,27 @@ Without such a plugin, all endpoints return ``501 Not Implemented``.
 
       {
         "status": "inProgress",
-        "x-medkit-phase": "preparing",
         "progress": 65,
         "sub_progress": [
           {"name": "download", "progress": 100},
           {"name": "verify", "progress": 30}
-        ]
+        ],
+        "x-medkit": {
+          "phase": "preparing"
+        }
       }
 
    **Status values:** ``pending``, ``inProgress``, ``completed``, ``failed``
 
    A successful ``POST /api/v1/updates`` seeds a ``pending`` status for the package,
-   so this endpoint returns ``200`` with ``{"status": "pending"}`` immediately after
-   registration, before any ``prepare`` or ``execute`` call.
+   so this endpoint returns ``200`` with ``{"status": "pending", "x-medkit": {"phase": "none"}}``
+   immediately after registration, before any ``prepare`` or ``execute`` call.
 
-   **Vendor extension ``x-medkit-phase``** (non-standard, SOVD-compatible):
+   **Vendor extension ``x-medkit.phase``** (non-standard, SOVD-compatible):
    ``none``, ``preparing``, ``prepared``, ``executing``, ``executed``,
    ``failed``, ``deleting``. Differentiates "prepare completed" (``status``
-   ``completed`` + ``x-medkit-phase`` ``prepared``) from "execute completed"
-   (``status`` ``completed`` + ``x-medkit-phase`` ``executed``). Clients that
+   ``completed`` + ``x-medkit.phase`` ``prepared``) from "execute completed"
+   (``status`` ``completed`` + ``x-medkit.phase`` ``executed``). Clients that
    only consume the standard ``status`` field continue to work unchanged.
 
    When ``status`` is ``failed``, an ``error`` object is included:

src/ros2_medkit_fault_manager/src/snapshot_capture.cpp

@@ -40,6 +40,18 @@ struct LockedSubscriptionGuard {
   rclcpp::GenericSubscription::SharedPtr subscription;
   rclcpp::CallbackGroup::SharedPtr callback_group;
 
+  // Explicit constructor instead of relying on aggregate initialization:
+  // C++20 [dcl.init.aggr] disqualifies a class as an aggregate if it has
+  // any user-declared constructors, including ``= delete`` ones (the rule
+  // tightened from "user-provided" in C++17 to "user-declared" in C++20).
+  // Rolling's gcc 14 / libstdc++ enforces the C++20 wording even when
+  // CMAKE_CXX_STANDARD is 17, so brace-init ``LockedSubscriptionGuard{
+  // &mtx, sub, cg}`` would fail to find a matching constructor on rolling.
+  // An explicit constructor sidesteps the aggregate-init rules entirely.
+  LockedSubscriptionGuard(std::mutex * m, rclcpp::GenericSubscription::SharedPtr s, rclcpp::CallbackGroup::SharedPtr cg)
+    : mtx(m), subscription(std::move(s)), callback_group(std::move(cg)) {
+  }
+
   ~LockedSubscriptionGuard() {
     if (!subscription && !callback_group) {
       return;

src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2_common/ros2_subscription_executor.hpp

@@ -189,17 +189,37 @@ class Ros2SubscriptionExecutor final {
    * disappearing, types changing). Use to invalidate per-topic caches or evict stale
    * pool entries.
    *
-   * @warning A registered callback must NOT call remove_graph_change() on its own
-   *          token from inside the callback - graph_mtx_ is non-recursive and a
-   *          self-removal attempt deadlocks. Drop the token through a separate task
-   *          posted to the worker if dynamic deregistration is needed.
+   * @warning DO NOT call remove_graph_change() on this token from inside the
+   *          callback body. The deadlock is concrete: remove_graph_change()
+   *          acquires graph_mtx_ and waits for graph_in_flight_[token] to
+   *          drain to zero, but the wrapper that decrements is the very call
+   *          frame that ran your callback. The worker thread is now blocked
+   *          on cv.wait, the decrement never runs, the wait never returns.
+   *          Symptom in production: hung worker thread, /health stops
+   *          updating, no log line. There is no runtime detection (would
+   *          need thread_local TLS, banned in code linked into the gateway
+   *          plugin MODULE; see CLAUDE.md). Workaround for dynamic
+   *          deregistration: have the callback hand the token to a
+   *          non-worker thread via a queue/promise, and call
+   *          remove_graph_change() from there.
    *
    * @return Opaque token in range [0, kMaxGraphListeners). `kMaxGraphListeners` if
    *         all slots are taken.
    */
   [[nodiscard]] std::size_t on_graph_change(GraphCallback cb);
 
-  /// Remove a previously-registered graph callback. Idempotent. See warning on on_graph_change.
+  /**
+   * @brief Remove a previously-registered graph callback. Idempotent.
+   *
+   * Synchronously drains in-flight execution: returns only after any worker
+   * thread that already entered this slot's callback has exited it. After
+   * remove_graph_change() returns, the caller can safely tear down state that
+   * the callback's captured `this` pointer references; the executor guarantees
+   * no further dispatch and no in-flight call.
+   *
+   * See warning on on_graph_change(): self-removal from inside the callback
+   * deadlocks.
+   */
   void remove_graph_change(std::size_t token);
 
   /// True after shutdown has started. Monotonic. Use to skip re-posting work on teardown.
@@ -269,10 +289,19 @@ class Ros2SubscriptionExecutor final {
   std::atomic<std::size_t> watchdog_trips_{0};
   std::atomic<std::size_t> graph_events_received_{0};
 
-  // Graph callbacks (pre-allocated, Tier 1)
+  // Graph callbacks (pre-allocated, Tier 1).
+  // graph_in_flight_[i] tracks how many worker invocations of slot i's
+  // callback are currently mid-execution. fire_graph_callbacks() increments
+  // it when snapshotting under graph_mtx_, the snapshot wrapper decrements
+  // it after the callback returns and notifies graph_in_flight_cv_. This
+  // lets remove_graph_change() block until in-flight execution drains, so a
+  // consumer's destructor can safely tear down state that the callback's
+  // captured `this` references.
   mutable std::mutex graph_mtx_;
   std::array<GraphCallback, kMaxGraphListeners> graph_callbacks_{};
   std::array<bool, kMaxGraphListeners> graph_slot_used_{};
+  std::array<int, kMaxGraphListeners> graph_in_flight_{};
+  std::condition_variable graph_in_flight_cv_;
 
   // Public rclcpp graph API - stable across Humble / Jazzy / Rolling
   rclcpp::Event::SharedPtr graph_event_;

src/ros2_medkit_gateway/include/ros2_medkit_gateway/updates/update_types.hpp

@@ -33,7 +33,7 @@ struct UpdateFilter {
 /// Status of an update operation (SOVD-compliant enum values)
 enum class UpdateStatus { Pending, InProgress, Completed, Failed };
 
-/// Lifecycle phase - exposed as SOVD vendor extension `x-medkit-phase`.
+/// Lifecycle phase - exposed as SOVD vendor extension `x-medkit.phase`.
 /// Differentiates "prepare completed" from "execute completed" which share
 /// status=completed in the SOVD standard enum.
 enum class UpdatePhase { None, Preparing, Prepared, Executing, Executed, Failed, Deleting };
@@ -94,7 +94,7 @@ class UpdateProgressReporter {
   std::mutex & mutex_;
 };
 
-/// Serialize UpdatePhase to its `x-medkit-phase` string value.
+/// Serialize UpdatePhase to its `x-medkit.phase` string value.
 inline const char * update_phase_to_string(UpdatePhase phase) {
   switch (phase) {
     case UpdatePhase::None:
@@ -116,7 +116,7 @@ inline const char * update_phase_to_string(UpdatePhase phase) {
 }
 
 /// Serialize UpdateStatusInfo to SOVD-compliant JSON.
-/// Adds vendor extension `x-medkit-phase` to distinguish prepare-completed from
+/// Adds vendor extension `x-medkit.phase` to distinguish prepare-completed from
 /// execute-completed (both report SOVD status=completed).
 inline nlohmann::json update_status_to_json(const UpdateStatusInfo & status) {
   nlohmann::json j;
@@ -134,7 +134,7 @@ inline nlohmann::json update_status_to_json(const UpdateStatusInfo & status) {
       j["status"] = "failed";
       break;
   }
-  j["x-medkit-phase"] = update_phase_to_string(status.phase);
+  j["x-medkit"] = {{"phase", update_phase_to_string(status.phase)}};
   if (status.progress.has_value()) {
     j["progress"] = *status.progress;
   }

src/ros2_medkit_gateway/src/http/handlers/config_handlers.cpp

@@ -311,15 +311,14 @@ void ConfigHandlers::handle_list_configurations(const httplib::Request & req, ht
 
             // Add source info for aggregated configurations
             if (agg_configs.is_aggregated) {
-              config_meta["x-medkit-source"] = node_info.app_id;
+              config_meta["x-medkit"] = {{"source", node_info.app_id}};
             }
 
             items.push_back(config_meta);
 
             // Also track full parameter info
             json param_with_source = param;
-            param_with_source["x-medkit-source"] = node_info.app_id;
-            param_with_source["x-medkit-node"] = node_info.node_fqn;
+            param_with_source["x-medkit"] = {{"source", node_info.app_id}, {"node", node_info.node_fqn}};
             all_parameters.push_back(param_with_source);
           }
         }

src/ros2_medkit_gateway/src/http/rest_server.cpp

@@ -1127,8 +1127,8 @@ void RESTServer::setup_routes() {
               })
           .tag("Discovery")
           .summary("Get app host component")
-          .description("Returns the component hosting this app.")
-          .response(200, "Host component", SB::ref("EntityDetail"))
+          .description("Returns the component hosting this app as a single-element collection.")
+          .response(200, "Host component(s)", SB::ref("EntityList"))
           .operation_id("getAppHost");
 
       reg.get(entity_path + "/depends-on",

src/ros2_medkit_gateway/src/openapi/schema_builder.cpp

@@ -160,12 +160,23 @@ nlohmann::json SchemaBuilder::items_wrapper(const nlohmann::json & item_schema)
 }
 
 nlohmann::json SchemaBuilder::configuration_metadata_schema() {
-  return {{"type", "object"},
+  return {
+      {"type", "object"},
+      {"properties",
+       {{"id", {{"type", "string"}, {"description", "Configuration parameter ID"}}},
+        {"name", {{"type", "string"}, {"description", "Parameter name"}}},
+        {"type", {{"type", "string"}, {"description", "Parameter type (e.g. 'parameter')"}}},
+        {"x-medkit",
+         {{"type", "object"},
+          {"description", "Vendor extensions (medkit)"},
           {"properties",
-           {{"id", {{"type", "string"}, {"description", "Configuration parameter ID"}}},
-            {"name", {{"type", "string"}, {"description", "Parameter name"}}},
-            {"type", {{"type", "string"}, {"description", "Parameter type (e.g. 'parameter')"}}}}},
-          {"required", {"id", "name", "type"}}};
+           {{"source",
+             {{"type", "string"},
+              {"description", "App ID that owns this parameter (only present in aggregated configurations)"}}},
+            {"node",
+             {{"type", "string"},
+              {"description", "Node FQN providing this parameter (only present in aggregated configurations)"}}}}}}}}},
+      {"required", {"id", "name", "type"}}};
 }
 
 nlohmann::json SchemaBuilder::configuration_read_value_schema() {
@@ -538,12 +549,31 @@ nlohmann::json SchemaBuilder::update_status_schema() {
       {"properties", {{"name", {{"type", "string"}}}, {"progress", {{"type", "number"}}}}},
       {"required", {"name", "progress"}}};
 
+  nlohmann::json x_medkit_schema = {
+      {"type", "object"},
+      {"description", "Vendor extensions (medkit)"},
+      {"properties",
+       {{"phase",
+         {{"type", "string"},
+          {"enum", {"none", "preparing", "prepared", "executing", "executed", "failed", "deleting"}},
+          {"description", "Internal lifecycle phase, distinguishes prepare-completed from execute-completed"}}}}},
+      {"required", {"phase"}}};
+
+  // x-medkit is optional in the SOVD payload (clients may ignore vendor
+  // extensions; matches the convention in fault_detail_schema and
+  // entity_detail_schema). When the gateway DOES emit the x-medkit object,
+  // however, ``phase`` is mandatory inside it - that scope is enforced by
+  // the inner ``required: {phase}`` above, NOT by listing x-medkit in the
+  // parent's required list. The drift test in test_openapi_response_drift
+  // covers regression on the emit side. If x-medkit is ever dropped from
+  // the parent properties, the inner required must be revisited too.
   return {{"type", "object"},
           {"properties",
            {{"status", {{"type", "string"}, {"enum", {"pending", "inProgress", "completed", "failed"}}}},
             {"progress", {{"type", "number"}}},
             {"sub_progress", {{"type", "array"}, {"items", sub_progress_schema}}},
-            {"error", {{"type", "string"}}}}},
+            {"error", {{"type", "string"}}},
+            {"x-medkit", x_medkit_schema}}},
           {"required", {"status"}}};
 }
 

src/ros2_medkit_gateway/src/ros2_common/ros2_subscription_executor.cpp

@@ -135,9 +135,17 @@ void Ros2SubscriptionExecutor::remove_graph_change(std::size_t token) {
   if (token >= kMaxGraphListeners) {
     return;
   }
-  std::lock_guard<std::mutex> lk(graph_mtx_);
+  std::unique_lock<std::mutex> lk(graph_mtx_);
   graph_slot_used_[token] = false;
   graph_callbacks_[token] = nullptr;
+  // Wait for any worker-thread invocation that already snapshotted this
+  // slot to finish executing the callback. Without this drain, a wrapper
+  // task posted by fire_graph_callbacks() could call cb() against a
+  // captured `this` after the consumer's destructor has freed members,
+  // producing the heap-use-after-free in the snapshotted callback body.
+  graph_in_flight_cv_.wait(lk, [this, token] {
+    return graph_in_flight_[token] == 0;
+  });
 }
 
 Ros2SubscriptionExecutor::Stats Ros2SubscriptionExecutor::stats() const {
@@ -309,30 +317,64 @@ void Ros2SubscriptionExecutor::fire_graph_callbacks() {
   // sweep, which can take 15 minutes by default. All-or-nothing delivery
   // means a single dropped post at most loses one whole graph event, which
   // graph_poll_tick() will refire on the next change.
-  std::vector<GraphCallback> snapshot;
+  //
+  // Each entry pairs the callback with its slot token: when the wrapper
+  // finishes invoking the callback, it decrements graph_in_flight_[token]
+  // so remove_graph_change() can synchronously wait for in-flight execution
+  // to drain.
+  struct PendingCb {
+    std::size_t token;
+    GraphCallback cb;
+  };
+  std::vector<PendingCb> snapshot;
   snapshot.reserve(kMaxGraphListeners);
   {
     std::lock_guard<std::mutex> lk(graph_mtx_);
     for (std::size_t i = 0; i < kMaxGraphListeners; ++i) {
       if (graph_slot_used_[i] && graph_callbacks_[i]) {
-        snapshot.push_back(graph_callbacks_[i]);
+        snapshot.push_back({i, graph_callbacks_[i]});
+        ++graph_in_flight_[i];
       }
     }
   }
   if (snapshot.empty()) {
     return;
   }
-  (void)post([cbs = std::move(snapshot)] {
-    for (const auto & cb : cbs) {
+  // Capture tokens before moving the snapshot into the wrapper so we can
+  // roll back graph_in_flight_ if post() rejects the wrapper (queue full
+  // or shutting down).
+  std::vector<std::size_t> reserved_tokens;
+  reserved_tokens.reserve(snapshot.size());
+  for (const auto & p : snapshot) {
+    reserved_tokens.push_back(p.token);
+  }
+  const bool posted = post([this, cbs = std::move(snapshot)] {
+    for (const auto & p : cbs) {
       try {
-        cb();
+        p.cb();
       } catch (const std::exception & ex) {
         RCLCPP_ERROR(rclcpp::get_logger("ros2_subscription_executor"), "Graph callback threw exception: %s", ex.what());
       } catch (...) {
         RCLCPP_ERROR(rclcpp::get_logger("ros2_subscription_executor"), "Graph callback threw unknown exception");
       }
+      {
+        std::lock_guard<std::mutex> lk(graph_mtx_);
+        --graph_in_flight_[p.token];
+      }
+      graph_in_flight_cv_.notify_all();
     }
   });
+  if (!posted) {
+    // Queue full or shutting down: the wrapper will not run, so we must
+    // decrement the counters we incremented to keep remove_graph_change()
+    // from waiting forever. snapshot is in moved-from state here; use
+    // reserved_tokens captured above the post() call.
+    std::lock_guard<std::mutex> lk(graph_mtx_);
+    for (auto t : reserved_tokens) {
+      --graph_in_flight_[t];
+    }
+    graph_in_flight_cv_.notify_all();
+  }
 }
 
 }  // namespace ros2_medkit_gateway::ros2_common

src/ros2_medkit_gateway/src/updates/update_manager.cpp

@@ -160,9 +160,9 @@ tl::expected<void, UpdateError> UpdateManager::delete_update(const std::string &
       if (it != states_.end() && it->second) {
         // Mark the package as failed consistently across all surfaces
         // exposed via GET /updates/{id}/status: the SOVD `status` field,
-        // the `x-medkit-phase` vendor extension, and `error_message`.
+        // the `x-medkit.phase` vendor extension, and `error_message`.
         // Without updating all three, clients see payloads like
-        // {status:pending, x-medkit-phase:failed} with no error context.
+        // {status:pending, "x-medkit":{phase:failed}} with no error context.
         it->second->status.status = UpdateStatus::Failed;
         it->second->status.phase = UpdatePhase::Failed;
         it->second->status.error_message = err.message;