fix(aggregation): target-filtered per-entity fan-out

Per-entity collection endpoints (/components/{id}/logs, /apps/{id}/data,
/areas/{id}/faults, /functions/{id}/logs, etc.) unconditionally fanned
out req.path to every healthy peer. Peers that do not host the entity
returned 404, the response was flagged as partial with those peers in
failed_peers, even though nothing is broken.

Fan-out is now directed at the peers that actually host or contribute to
the entity:

- AggregationManager tracks a peer_contributors_by_entity_ map alongside
  the existing routing_table_. gateway_node rebuilds both after every
  discovery cycle by walking contributors on the merged entities and
  stripping the 'peer:' prefix.
- get_peer_contributors(id) unions the routing-table owner (routed leaf,
  collision-renamed peer-only entity) with the per-entity contributor
  map (merged Areas/Functions, hierarchical parent Components that were
  stripped from the routing table during classification).
- fan_out_get accepts an optional target_peers list; when supplied, only
  matching healthy peers are queried. Non-contributors never see the
  request, so they cannot appear in failed_peers.
- merge_peer_items extracts the entity id from per-entity paths, asks
  AggregationManager for contributors, and passes the list as the
  target filter. Local-only entities produce an empty list and skip
  fan-out entirely. Global collection endpoints without an entity id
  keep fan-out-to-all behavior.

Covers the common cases: local-only entity (no fan-out), routed leaf
(only the owning peer queried), merged/hierarchical entity (only
contributing peers queried), partial failure isolated to the specific
contributor that failed. Updates aggregation.rst design doc and extends
unit tests for AggregationManager (has_peer_contributors,
get_peer_contributors, empty-entry pruning) and fan-out helpers (owner-
only, contributor-only, partial-only-on-contributor-failure).

Author: bburda

src/ros2_medkit_gateway/design/aggregation.rst

@@ -424,6 +424,33 @@ peers fail, the response body includes ``x-medkit.partial: true`` and
 ``X-Medkit-No-Fan-Out`` header to prevent recursive loops when peers have
 bidirectional aggregation.
 
+**Target-filtered fan-out.** For per-entity paths, ``merge_peer_items()``
+asks ``AggregationManager::get_peer_contributors(id)`` for the list of
+peers that host or contribute to the entity, and passes it as a filter to
+``fan_out_get()``. Requests reach only those peers; non-contributors are
+never queried so they cannot appear in ``failed_peers``. The set unions:
+
+- The routing table (remote leaves, collision-renamed peer-only entities).
+- A ``peer_contributors_by_entity_`` map maintained alongside the routing
+  table. ``gateway_node`` rebuilds both after every discovery cycle by
+  walking ``contributors`` on the merged Areas/Components/Apps/Functions,
+  stripping the ``"peer:"`` prefix and accumulating peer names per id.
+  Merged Areas/Functions with ID collisions and hierarchical parent
+  Components - both deliberately stripped from the routing table - still
+  reach their peers through this map.
+
+When the resolved list is empty (local-only entity), fan-out is skipped:
+no peer hosts the entity, so hitting peers would only produce spurious
+``partial: true`` / ``failed_peers``. Global endpoints (paths with no
+entity id, e.g. ``GET /api/v1/faults``) pass a ``nullptr`` filter and keep
+fan-out-to-all-healthy behavior.
+
+Entities freshly announced on a peer but not yet reflected in the local
+routing/contributor tables (a brief window between discovery cycles) are
+treated as local-only: their per-entity fan-out is deferred until the
+next cycle rebuilds the tables. This is a deliberate trade-off against
+re-enabling the spurious ``partial: true`` path.
+
 .. warning::
 
    Fan-out is synchronous on the httplib handler thread. Each request blocks

src/ros2_medkit_gateway/include/ros2_medkit_gateway/aggregation/aggregation_manager.hpp

@@ -203,6 +203,43 @@ class AggregationManager {
    */
   std::optional<std::string> find_peer_for_entity(const std::string & entity_id) const;
 
+  /**
+   * @brief Replace the map of per-entity peer contributors.
+   *
+   * Covers both routed leaf entities (entity fully hosted by a peer - also
+   * tracked via the routing table) and merged/hierarchical entities (served
+   * locally with contributions from one or more peers - e.g. a Component
+   * aggregating subcomponents from multiple gateways, or an Area/Function
+   * with an ID shared across peers).
+   *
+   * Fan-out helpers consult this map to target per-entity collection
+   * requests only at the peers that actually host / contribute to a given
+   * entity, avoiding spurious 404s from non-contributing peers.
+   *
+   * @param contributors New map: entity id -> list of peer names that
+   *        contribute to it. Entries with empty lists are discarded.
+   */
+  void update_peer_contributors(std::unordered_map<std::string, std::vector<std::string>> contributors);
+
+  /**
+   * @brief Check whether an entity id has at least one peer contributor.
+   *
+   * Returns true when the entity is either routed to a peer (remote leaf) or
+   * served locally but aggregated from peers (hierarchical / merged). Returns
+   * false for local-only entities and for unknown IDs. Thread-safe.
+   */
+  bool has_peer_contributors(const std::string & entity_id) const;
+
+  /**
+   * @brief List the peers that host / contribute to a given entity.
+   *
+   * Combines information from the routing table (routed leaves) and the
+   * per-entity contributors map (merged / hierarchical entities). Duplicates
+   * are removed while preserving first-seen order. Returns an empty vector
+   * for local-only entities and unknown IDs. Thread-safe.
+   */
+  std::vector<std::string> get_peer_contributors(const std::string & entity_id) const;
+
   /**
    * @brief Get the URL for a known peer by name
    * @param peer_name Name of the peer
@@ -223,17 +260,22 @@ class AggregationManager {
   void forward_request(const std::string & peer_name, const httplib::Request & req, httplib::Response & res);
 
   /**
-   * @brief Fan-out a GET request to all healthy peers in parallel
+   * @brief Fan-out a GET request to healthy peers in parallel.
    *
-   * Sends GET requests to all healthy peers concurrently via std::async,
-   * merges the "items" arrays from their responses. Returns partial results
-   * if some peers fail.
+   * Sends GET requests concurrently via std::async, merges the "items"
+   * arrays from their responses. Returns partial results if some peers fail.
    *
    * @param path Request path (e.g., "/api/v1/components")
    * @param auth_header Authorization header value (empty to omit)
-   * @return FanOutResult with merged items and failure info
+   * @param target_peers When non-null, only peers whose name is in this
+   *        list AND are currently healthy are queried. A non-null but empty
+   *        list means "no matching peers" and produces an empty result with
+   *        no fan-out. When null (default), all healthy peers are queried,
+   *        preserving the pre-filtering behavior for global endpoints.
+   * @return FanOutResult with merged items and failure info.
    */
-  FanOutResult fan_out_get(const std::string & path, const std::string & auth_header);
+  FanOutResult fan_out_get(const std::string & path, const std::string & auth_header,
+                           const std::vector<std::string> * target_peers = nullptr);
 
   /**
    * @brief Get peer status for /health endpoint
@@ -248,6 +290,7 @@ class AggregationManager {
   mutable std::shared_mutex mutex_;  // Declared before data it protects (destruction order)
   std::vector<std::shared_ptr<PeerClient>> peers_;
   std::unordered_map<std::string, std::string> routing_table_;
+  std::unordered_map<std::string, std::vector<std::string>> peer_contributors_by_entity_;
   std::vector<LeafCollisionWarning> leaf_warnings_;
 
   /**

src/ros2_medkit_gateway/include/ros2_medkit_gateway/http/fan_out_helpers.hpp

@@ -14,7 +14,10 @@
 
 #pragma once
 
+#include <array>
+#include <optional>
 #include <string>
+#include <string_view>
 
 #include <httplib.h>
 #include <nlohmann/json.hpp>
@@ -45,6 +48,43 @@ inline std::string url_encode_param(const std::string & value) {
   return result;
 }
 
+// Extract the entity id from a per-entity collection path like
+// `/api/v1/components/<id>/logs` or `/api/v1/apps/<id>/data/<sub>`.
+// Matching is intentionally permissive: any path containing
+// `/components|apps|areas|functions/<id>/<anything>` is treated as
+// per-entity, regardless of what the segment after `<id>` is. This keeps
+// future per-entity sub-endpoints eligible for target-filtered fan-out
+// without requiring a whitelist bump every time.
+//
+// Returns std::nullopt for:
+//   - global endpoints (`/api/v1/faults`, `/api/v1/health`, ...),
+//   - entity-detail endpoints (`/api/v1/components/<id>` with no trailing
+//     `/...` - these go through the forwarding path, not fan-out),
+//   - malformed paths with an empty entity segment (`/components//logs`).
+inline std::optional<std::string> extract_entity_id_for_fan_out(const std::string & path) {
+  static constexpr std::array<std::string_view, 4> kEntityCollections = {"/components/", "/apps/", "/areas/",
+                                                                         "/functions/"};
+  for (const auto & prefix : kEntityCollections) {
+    auto start = path.find(prefix);
+    if (start == std::string::npos) {
+      continue;
+    }
+    start += prefix.size();
+    auto end = path.find('/', start);
+    if (end == std::string::npos) {
+      // Entity detail endpoint like `/components/<id>` (no trailing collection).
+      // These are routed to peers via a different mechanism; not a fan-out case.
+      return std::nullopt;
+    }
+    std::string id = path.substr(start, end - start);
+    if (id.empty()) {
+      return std::nullopt;
+    }
+    return id;
+  }
+  return std::nullopt;
+}
+
 inline std::string build_fan_out_path(const httplib::Request & req) {
   if (req.params.empty()) {
     return req.path;
@@ -76,8 +116,23 @@ inline void merge_peer_items(AggregationManager * agg, const httplib::Request &
   if (agg->healthy_peer_count() == 0) {
     return;
   }
+  // For per-entity collection paths, target only the peers that host or
+  // contribute to the entity (routed leaves and merged / hierarchical
+  // entities). Local-only entities produce an empty target list and skip
+  // fan-out entirely, avoiding spurious `partial: true` / `failed_peers`
+  // from peers that do not own the entity. Global collection endpoints
+  // (paths without an entity id) keep fan-out-to-all behavior.
+  std::optional<std::vector<std::string>> contributors_buffer;
+  const std::vector<std::string> * target_peers = nullptr;
+  if (auto entity_id = extract_entity_id_for_fan_out(req.path); entity_id.has_value()) {
+    contributors_buffer = agg->get_peer_contributors(*entity_id);
+    if (contributors_buffer->empty()) {
+      return;  // local-only: no peer hosts this entity
+    }
+    target_peers = &contributors_buffer.value();
+  }
   auto fan_path = build_fan_out_path(req);
-  auto fan_result = agg->fan_out_get(fan_path, req.get_header_value("Authorization"));
+  auto fan_result = agg->fan_out_get(fan_path, req.get_header_value("Authorization"), target_peers);
   if (fan_result.merged_items.is_array() && !fan_result.merged_items.empty()) {
     if (!result.contains("items") || !result["items"].is_array()) {
       result["items"] = nlohmann::json::array();

src/ros2_medkit_gateway/src/aggregation/aggregation_manager.cpp

@@ -547,6 +547,50 @@ std::optional<std::string> AggregationManager::find_peer_for_entity(const std::s
   return std::nullopt;
 }
 
+void AggregationManager::update_peer_contributors(
+    std::unordered_map<std::string, std::vector<std::string>> contributors) {
+  // Drop entries with empty contributor lists - they would be indistinguishable
+  // from "local-only" lookups and only waste space.
+  for (auto it = contributors.begin(); it != contributors.end();) {
+    if (it->second.empty()) {
+      it = contributors.erase(it);
+    } else {
+      ++it;
+    }
+  }
+  std::unique_lock<std::shared_mutex> lock(mutex_);
+  peer_contributors_by_entity_ = std::move(contributors);
+}
+
+bool AggregationManager::has_peer_contributors(const std::string & entity_id) const {
+  std::shared_lock<std::shared_mutex> lock(mutex_);
+  if (routing_table_.count(entity_id) > 0u) {
+    return true;
+  }
+  auto it = peer_contributors_by_entity_.find(entity_id);
+  return it != peer_contributors_by_entity_.end() && !it->second.empty();
+}
+
+std::vector<std::string> AggregationManager::get_peer_contributors(const std::string & entity_id) const {
+  std::shared_lock<std::shared_mutex> lock(mutex_);
+  std::vector<std::string> result;
+  // Routed-leaf entry wins ordering: it always reflects the authoritative
+  // owner of the entity's runtime state.
+  auto rt_it = routing_table_.find(entity_id);
+  if (rt_it != routing_table_.end()) {
+    result.push_back(rt_it->second);
+  }
+  auto pc_it = peer_contributors_by_entity_.find(entity_id);
+  if (pc_it != peer_contributors_by_entity_.end()) {
+    for (const auto & name : pc_it->second) {
+      if (std::find(result.begin(), result.end(), name) == result.end()) {
+        result.push_back(name);
+      }
+    }
+  }
+  return result;
+}
+
 std::string AggregationManager::get_peer_url(const std::string & peer_name) const {
   std::shared_lock<std::shared_mutex> lock(mutex_);
 
@@ -609,22 +653,39 @@ void AggregationManager::forward_request(const std::string & peer_name, const ht
 }
 
 AggregationManager::FanOutResult AggregationManager::fan_out_get(const std::string & path,
-                                                                 const std::string & auth_header) {
+                                                                 const std::string & auth_header,
+                                                                 const std::vector<std::string> * target_peers) {
   // Per-peer result collected by each async task
   struct PeerResult {
     std::string peer_name;
     bool success{false};
     nlohmann::json items = nlohmann::json::array();
   };
 
+  // A non-null, empty target list means "no peers match" - skip fan-out
+  // entirely so callers don't see spurious partial flags. Keep the
+  // merged_items invariant (always a JSON array, never null) for downstream
+  // .is_array() checks.
+  if (target_peers != nullptr && target_peers->empty()) {
+    FanOutResult empty_result;
+    empty_result.merged_items = nlohmann::json::array();
+    return empty_result;
+  }
+
   // Snapshot healthy peers under lock, release before network I/O.
+  // When target_peers is non-null, restrict the snapshot to matching peers.
   std::vector<std::shared_ptr<PeerClient>> snapshot;
   {
     std::shared_lock<std::shared_mutex> lock(mutex_);
     for (const auto & peer : peers_) {
-      if (peer->is_healthy()) {
-        snapshot.push_back(peer);
+      if (!peer->is_healthy()) {
+        continue;
       }
+      if (target_peers != nullptr &&
+          std::find(target_peers->begin(), target_peers->end(), peer->name()) == target_peers->end()) {
+        continue;
+      }
+      snapshot.push_back(peer);
     }
   }
 

src/ros2_medkit_gateway/src/gateway_node.cpp

@@ -20,7 +20,10 @@
 #include <cinttypes>
 #include <mutex>
 #include <set>
+#include <string_view>
+#include <unordered_map>
 #include <unordered_set>
+#include <vector>
 
 #include "ros2_medkit_gateway/aggregation/network_utils.hpp"
 #include "ros2_medkit_gateway/entity_validation.hpp"
@@ -1682,6 +1685,41 @@ void GatewayNode::refresh_cache() {
       peer_routing_table = std::move(merged.routing_table);
       aggregation_mgr_->update_routing_table(peer_routing_table);
       aggregation_mgr_->set_leaf_warnings(std::move(merged.leaf_warnings));
+
+      // Track per-entity peer contributors: entity id -> list of peer
+      // names that host or contribute to it. Covers both routed leaves
+      // (also tracked via routing_table) and merged / hierarchical entities
+      // (Areas, Functions, parent Components). Fan-out helpers use this map
+      // to target per-entity collection requests at the peers that actually
+      // own the entity, avoiding spurious 404s from non-contributing peers.
+      //
+      // The "peer:" prefix is stripped from the contributors list produced
+      // by EntityMerger. Entries starting with anything else (including
+      // "local") are ignored.
+      std::unordered_map<std::string, std::vector<std::string>> peer_contributors;
+      static constexpr std::string_view kPeerPrefix = "peer:";
+      auto collect_contributors = [&peer_contributors](const auto & entities) {
+        for (const auto & e : entities) {
+          for (const auto & c : e.contributors) {
+            if (c.rfind(kPeerPrefix, 0) != 0) {
+              continue;
+            }
+            std::string peer_name = c.substr(kPeerPrefix.size());
+            if (peer_name.empty()) {
+              continue;
+            }
+            auto & list = peer_contributors[e.id];
+            if (std::find(list.begin(), list.end(), peer_name) == list.end()) {
+              list.push_back(std::move(peer_name));
+            }
+          }
+        }
+      };
+      collect_contributors(areas);
+      collect_contributors(all_components);
+      collect_contributors(apps);
+      collect_contributors(functions);
+      aggregation_mgr_->update_peer_contributors(std::move(peer_contributors));
     }
 
     // Inject plugin entities (non-hybrid) and refresh entity ownership (all modes).

src/ros2_medkit_gateway/src/updates/update_manager.cpp

@@ -153,17 +153,24 @@ tl::expected<void, UpdateError> UpdateManager::delete_update(const std::string &
     states_.erase(id);
   } else {
     // Rollback sentinel on failure
+    const auto & err = result.error();
     std::lock_guard<std::mutex> lock(mutex_);
     if (had_state) {
       auto it = states_.find(id);
       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`.
+        // Without updating all three, clients see payloads like
+        // {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;
       }
     } else {
       // Remove the sentinel we created - package never had state before
       states_.erase(id);
     }
-    const auto & err = result.error();
     switch (err.code) {
       case UpdateBackendError::NotFound:
         return tl::make_unexpected(UpdateError{UpdateErrorCode::NotFound, err.message});