gateway: answer a peer's GET /faults without fanning out again

docs/api/rest.rst

@@ -45,21 +45,34 @@ operations that read them.
 ``X-Medkit-No-Fan-Out``
    Answer from this gateway alone: do not query aggregated peers and do not
    merge their items. Read by the **per-entity** resource-collection list
-   endpoints (data, operations, configurations, faults, logs) and by ``GET
-   /api/v1/version-info`` - exactly the operations that declare it in the
-   OpenAPI document.
+   endpoints (data, operations, configurations, faults, logs), by the global
+   ``GET /api/v1/faults`` and by ``GET /api/v1/version-info`` - exactly the
+   operations that declare it in the OpenAPI document.
 
    The gateway sets it on its own outbound peer requests, which stops
    bidirectional aggregation from recursing **on the routes that check it**.
-   The global ``GET /api/v1/faults`` does not check it, so it neither declares
-   the header nor honours it (see :ref:`the fan-out design note
-   <aggregation-fan-out>`).
 
    **Presence-only.** The value is never read, so ``X-Medkit-No-Fan-Out:
    false`` suppresses fan-out exactly like any other value. The OpenAPI schema
    is a string rather than a boolean for that reason. Omit the header to get
    the aggregated answer.
 
+``X-Medkit-Fan-Out-Visited``
+   Read by the global ``GET /api/v1/faults`` only, and set by the gateway on
+   its own fan-out of that route: the instance ids of the gateways the request
+   has already passed through, comma-separated. Each aggregating gateway
+   publishes its id as ``x-medkit-instance-id`` on ``GET /api/v1/health``,
+   where its peers read it. A gateway does not ask a peer whose id is on the
+   path. A gateway asked anyway answers an empty list, because its faults are
+   already in the answer. Any other gateway answers with its own faults and
+   those of its peers, adding its id. This is what lets the global list reach
+   the end of a chain (A -> B -> C) while two gateways that aggregate each
+   other do not ask each other twice. It takes precedence over
+   ``X-Medkit-No-Fan-Out``, which the same request also carries. A path of 16
+   ids is not extended: the gateway answers for itself and reports its peers
+   as ``too-deep``. Clients do not send it (see :ref:`the fan-out design note
+   <aggregation-fan-out>`).
+
 Server Capabilities
 -------------------
 
@@ -3892,6 +3905,13 @@ why, one entry per name, with ``reason`` one of:
    * - ``invalid-response``
      - The peer answered with something that is not the JSON the route
        promises.
+   * - ``too-deep``
+     - Global ``GET /faults`` only. The request had already passed through 16
+       gateways, so this gateway did not ask the peer.
+
+On the global ``GET /faults`` a peer that could not reach one of its own
+peers is reported partial in turn. The entry names both hops,
+``<peer>/<its peer>``, with the reason the peer gave.
 
 Without ``peer_failures`` a client cannot tell a busy subsystem from a dead
 one, which is the whole of what a partial answer is asked.

docs/config/aggregation.rst

@@ -491,6 +491,14 @@ into the local response. This applies to:
 - ``GET /api/v1/{entity_type}/{id}/logs``
 - ``GET /api/v1/faults`` (global fault list)
 
+The global fault list reaches the whole chain: in A -> B -> C, ``GET
+/api/v1/faults`` on A lists the faults of A, B and C. Each gateway adds its own
+id to an ``X-Medkit-Fan-Out-Visited`` header on the peer requests and does not
+ask a peer whose id is already there. Two gateways that list each other as
+peers therefore do not pass the request back and forth. If B cannot reach C,
+A's answer is partial and names ``subsystem_b/subsystem_c`` in
+``failed_peers``.
+
 If a peer request fails during fan-out (peer unreachable or non-2xx response),
 the response body includes:
 

src/ros2_medkit_gateway/design/aggregation.rst

@@ -621,26 +621,54 @@ healthy peers, collects the responses, and merges the ``items`` arrays. If some
 peers fail, the response body includes ``x-medkit.partial: true`` and
 ``x-medkit.failed_peers``. Fan-out requests carry an ``X-Medkit-No-Fan-Out``
 header, and both helpers return early when the incoming request has it, so a
-peer that aggregates back never fans out a second time. These are the routes
-that declare the header in the OpenAPI document
-(``RouteEntry::fan_out_aware()``).
-
-.. warning::
-
-   The global ``GET /api/v1/faults`` is **not** one of them.
-   ``FaultHandlers::list_all_faults`` calls ``fan_out_get()`` directly rather
-   than through either helper, so it never inspects ``X-Medkit-No-Fan-Out`` -
-   and ``fan_out_get()`` is the code that *sets* the header outbound. Nothing
-   else in ``aggregation/`` guards the loop. Two gateways that peer with each
-   other therefore recurse unbounded on this one route: A queries B, B queries
-   A, and each hop holds a ``std::async`` thread until its timeout, so the
-   thread cost grows with recursion depth.
-
-   The route consequently does not declare the header either - advertising an
-   opt-out it ignores would be worse than silence. Fixing this means routing
-   ``list_all_faults`` through the helper (which changes the per-item wire
-   shape it deliberately preserves) or adding a loop guard inside
-   ``fan_out_get()``; both are aggregation changes, not documentation ones.
+peer that aggregates back never fans out a second time. These routes and the
+global fault list below are the ones that declare the header in the OpenAPI
+document (``RouteEntry::fan_out_aware()``).
+
+**The global fault list** (``GET /api/v1/faults``) fans out as well, but it
+cannot stop at the first peer. There is no routing table behind it: in a chain
+A -> B -> C the only way A lists C's faults is for B to ask C. So
+``FaultHandlers::list_all_faults`` (and ``PluginContext::list_all_faults``,
+which calls the same route on the peers) sends ``X-Medkit-Fan-Out-Visited``
+next to ``X-Medkit-No-Fan-Out``. The header carries the instance ids of the
+gateways the request has passed through. Each ``AggregationManager`` generates
+its id at startup, because a peer name is what the aggregator calls a peer and
+a gateway cannot recognise itself by it. It publishes the id as
+``x-medkit-instance-id`` on ``/health``, and ``PeerClient::check_health``
+records each peer's. ``plan_global_fault_fan_out()`` in ``fan_out_helpers.hpp``
+reads the request:
+
+- No visited path and no ``X-Medkit-No-Fan-Out``: a client request. Answer
+  locally and fan out with the path set to this gateway's id.
+- Visited path that does not name this gateway: answer locally and fan out
+  with this gateway's id added. ``X-Medkit-No-Fan-Out`` is ignored here,
+  because ``fan_out_get()`` puts it on every peer request.
+- Visited path that names this gateway: the request came back round a loop.
+  Answer ``200`` with an empty list, since this gateway's faults are already
+  in the answer the first pass is building.
+- A path of 16 ids, or one holding an id longer than 64 characters: answer
+  locally, do not extend it, and report every healthy peer as failed with
+  reason ``too-deep``, so a truncated list is not reported complete.
+- ``X-Medkit-No-Fan-Out`` alone: answer locally only. That is a client opting
+  out, or a gateway that predates the visited path.
+
+``fan_out_get()`` does not send the request to a peer whose recorded id is on
+the path. Two gateways that aggregate each other therefore do not ask each
+other twice: A asks B with ``[A]``, and B asks only its other peers. This
+matters for more than latency. A is holding a worker while it waits for B, so
+a request sent back to A could only be answered once another of A's workers
+is free. Under as many concurrent requests as A has workers, none would be,
+and every one would wait for the peer timeout. The empty answer for a gateway
+that finds itself on the path covers a peer whose id is not known yet (before
+its first health check) or one that predates the id.
+
+A peer that answers partial makes the result partial too. Its failed peers are
+named ``<peer>/<its peer>`` with the reason it gave, so a dead gateway at the
+end of a chain shows up at the head. ``x-medkit.count`` is the length of the
+merged ``items``.
+
+The path does not deduplicate a gateway reached over two different routes in
+the same request (A -> B -> C plus A -> C): C answers both, as it did before.
 
 **Target-filtered fan-out.** For per-entity paths, ``merge_peer_items()``
 asks ``AggregationManager::get_peer_contributors(id)`` for the list of

src/ros2_medkit_gateway/design/dto_contract.rst

@@ -714,7 +714,8 @@ Further ``RouteEntry`` knobs shape the published operation:
 - ``fan_out_aware()`` - this route reads the ``X-Medkit-No-Fan-Out`` request
   header, i.e. a client can ask it to answer from this gateway alone instead of
   merging aggregated peers. Carried by the routes whose handlers go through
-  ``fan_out_collection`` or ``merge_peer_items``. The declared schema is a bare
+  ``fan_out_collection`` or ``merge_peer_items``, and by the global ``GET
+  /faults``, which reads it through ``plan_global_fault_fan_out``. The declared schema is a bare
   string, not a boolean: the gateway tests ``has_header`` and never reads the
   value, so ``X-Medkit-No-Fan-Out: false`` still suppresses fan-out and a
   boolean schema would promise a generated client the opposite. Also

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

@@ -32,6 +32,31 @@
 
 namespace ros2_medkit_gateway {
 
+/// Instance ids of the gateways a global `GET /faults` fan-out has already
+/// passed through, comma-separated. A gateway that finds its own id here has
+/// its faults in the answer already; one that does not answers for itself and
+/// for its own peers. See `plan_global_fault_fan_out`.
+inline constexpr const char * kFanOutVisitedHeader = "X-Medkit-Fan-Out-Visited";
+
+/// The ids on an `X-Medkit-Fan-Out-Visited` value, trimmed, empty entries dropped.
+inline std::vector<std::string> split_fan_out_visited(const std::string & value) {
+  std::vector<std::string> ids;
+  for (size_t start = 0; start <= value.size();) {
+    auto end = value.find(',', start);
+    if (end == std::string::npos) {
+      end = value.size();
+    }
+    std::string id = value.substr(start, end - start);
+    id.erase(0, id.find_first_not_of(" \t"));
+    id.erase(id.find_last_not_of(" \t") + 1);
+    if (!id.empty()) {
+      ids.push_back(std::move(id));
+    }
+    start = end + 1;
+  }
+  return ids;
+}
+
 /// A peer's name and where to reach it.
 struct PeerEndpoint {
   std::string name;
@@ -193,6 +218,14 @@ class AggregationManager {
   /// Get the number of known peers (static + discovered)
   size_t peer_count() const;
 
+  /// Random id of this gateway process, put on `X-Medkit-Fan-Out-Visited`.
+  ///
+  /// Generated, not configured: peer names are what the aggregator calls a
+  /// peer, so a gateway cannot recognise itself by one.
+  const std::string & instance_id() const {
+    return instance_id_;
+  }
+
   /**
    * @brief Add a dynamically discovered peer
    *
@@ -392,10 +425,18 @@ class AggregationManager {
    *        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.
+   * @param extra_headers Sent on every peer request next to
+   *        `X-Medkit-No-Fan-Out`. When they carry `X-Medkit-Fan-Out-Visited`,
+   *        a peer whose instance id is on that path is not asked at all: its
+   *        answer is already counted, and asking it would hold a worker on
+   *        both sides until it answered empty. A peer that reports itself
+   *        partial makes this result partial too, with its failed peers named
+   *        `<peer>/<its peer>`.
    * @return FanOutResult with merged items and failure info.
    */
   FanOutResult fan_out_get(const std::string & path, const std::string & auth_header,
-                           const std::vector<std::string> * target_peers = nullptr);
+                           const std::vector<std::string> * target_peers = nullptr,
+                           const httplib::Headers & extra_headers = {});
 
   /**
    * @brief Get peer status for /health endpoint
@@ -406,6 +447,7 @@ class AggregationManager {
  private:
   AggregationConfig config_;
   log::Logger logger_;
+  std::string instance_id_;
   size_t static_peer_count_{0};  ///< Number of statically configured peers (not subject to max_discovered_peers)
   /// Whether a peer whose picture could not be refreshed can still be reached.
   ///

src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/aggregation/peer_client.hpp

@@ -178,11 +178,16 @@ class PeerClient {
   /// Check if the peer was healthy at last health check
   bool is_healthy() const;
 
+  /// The peer's `x-medkit-instance-id` from its last health check, empty if it
+  /// sent none. Lets a fan-out skip a peer already on the visited path.
+  std::string instance_id() const;
+
   /**
    * @brief Perform a health check against the peer
    *
    * GETs /api/v1/health on the peer. Sets the internal healthy_ flag
-   * based on whether a 200 response was received.
+   * based on whether a 200 response was received, and records the peer's
+   * instance id.
    */
   void check_health();
 
@@ -308,6 +313,9 @@ class PeerClient {
   std::atomic<bool> healthy_{false};
   std::atomic<bool> shutdown_requested_{false};
 
+  mutable std::mutex instance_id_mutex_;
+  std::string instance_id_;
+
   std::mutex client_mutex_;
   std::unique_ptr<httplib::Client> client_;
 

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

@@ -14,7 +14,9 @@
 
 #pragma once
 
+#include <algorithm>
 #include <array>
+#include <cstddef>
 #include <optional>
 #include <string>
 #include <string_view>
@@ -111,6 +113,69 @@ inline std::string build_fan_out_path(const httplib::Request & req) {
   return path;
 }
 
+/// How a gateway answers the global `GET /faults`.
+///
+/// That list has no routing table behind it: the only way to list the faults
+/// of a gateway two hops away is for the hop in between to fan out again. So
+/// the route does not stop at `X-Medkit-No-Fan-Out` the way the per-entity
+/// collections do. Each gateway adds its instance id to
+/// `X-Medkit-Fan-Out-Visited` instead. A peer whose id is on the path is not
+/// asked, and one asked anyway (its id not yet learned from its health check)
+/// answers an empty list. A chain A -> B -> C still reaches C, and two gateways
+/// that aggregate each other do not ask each other twice.
+struct GlobalFaultFanOut {
+  enum class Mode {
+    kFanOut,          ///< Answer locally and ask the peers, sending `visited`.
+    kLocalOnly,       ///< Answer from this gateway's fault manager only.
+    kAlreadyCounted,  ///< This gateway is on the path: answer an empty list.
+    kTooDeep,         ///< The path cannot be extended: answer locally, report the peers not asked.
+  };
+  Mode mode{Mode::kLocalOnly};
+  std::string visited;
+};
+
+/// Longest `X-Medkit-Fan-Out-Visited` path that is still extended.
+inline constexpr size_t kMaxFanOutVisited = 16;
+/// Longest instance id accepted on that path. Ours are 32 hex digits.
+inline constexpr size_t kMaxFanOutVisitedIdLength = 64;
+
+/// `self_id` is `AggregationManager::instance_id()`, empty when aggregation is off.
+///
+/// `X-Medkit-No-Fan-Out` alone keeps its meaning: answer from this gateway
+/// alone. That is what a client opting out sends, and what a gateway that
+/// predates the visited path sends. Next to the visited path it is ignored,
+/// because `fan_out_get` puts it on every peer request.
+inline GlobalFaultFanOut plan_global_fault_fan_out(const httplib::Request & req, const std::string & self_id) {
+  using Mode = GlobalFaultFanOut::Mode;
+  if (self_id.empty()) {
+    return {Mode::kLocalOnly, {}};
+  }
+  if (!req.has_header(kFanOutVisitedHeader)) {
+    if (req.has_header("X-Medkit-No-Fan-Out")) {
+      return {Mode::kLocalOnly, {}};
+    }
+    return {Mode::kFanOut, self_id};
+  }
+  const auto ids = split_fan_out_visited(req.get_header_value(kFanOutVisitedHeader));
+  if (std::find(ids.begin(), ids.end(), self_id) != ids.end()) {
+    return {Mode::kAlreadyCounted, {}};
+  }
+  // A path this gateway cannot extend still gets its own faults: answering
+  // for itself is right, asking further is not.
+  bool readable = ids.size() < kMaxFanOutVisited;
+  std::string visited;
+  for (const auto & id : ids) {
+    readable = readable && id.size() <= kMaxFanOutVisitedIdLength;
+    visited += id;
+    visited += ',';
+  }
+  if (!readable) {
+    return {Mode::kTooDeep, {}};
+  }
+  visited += self_id;
+  return {Mode::kFanOut, std::move(visited)};
+}
+
 /// Fan-out GET to peer gateways and merge their items into `result["items"]`.
 /// Aggregation metadata (partial, failed_peers) is written into `ext_json`
 /// (a plain JSON object that callers fold into the response x-medkit block).

src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/aggregation.hpp

@@ -64,9 +64,10 @@ inline constexpr std::string_view dto_name<DroppedItem> = "DroppedItem";
 // tell a slow subsystem from a dead one.
 //
 // Wire keys:
-//   peer   - peer name, matching the entry in failed_peers
+//   peer   - peer name, matching the entry in failed_peers. On the global
+//            fault list a peer's own failed peer is named <peer>/<its peer>.
 //   reason - one of: timeout, unreachable, canceled, error-status, too-large,
-//            invalid-response
+//            invalid-response, too-deep (global fault list only)
 // ---------------------------------------------------------------------------
 struct PeerFailure {
   std::string peer;

src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/health.hpp

@@ -172,6 +172,10 @@ struct Health {
   /// why a peer refused their stream with 503.
   std::optional<nlohmann::json> x_medkit_sse;
   std::optional<nlohmann::json> peers;  // free-form array of peer status objects
+  /// Random id of this gateway process, present when aggregation is on. Wire
+  /// key: "x-medkit-instance-id". An aggregating peer reads it in its health
+  /// check, so a global fault-list fan-out can skip a gateway it already passed.
+  std::optional<std::string> x_medkit_instance_id;
   // NOT optional: these two are emitted in every mode, aggregation or not, so
   // the generated schema must list them in `required` and let a typed client
   // read them without a presence check. An empty `warnings` array is the
@@ -190,8 +194,12 @@ inline constexpr auto dto_fields<Health> = std::make_tuple(
     field("discovery", &Health::discovery), field("x-medkit-data-provider", &Health::x_medkit_data_provider),
     field("x-medkit-subscription-executor", &Health::x_medkit_subscription_executor),
     field("x-medkit-entity-cache", &Health::x_medkit_entity_cache), field("x-medkit-sse", &Health::x_medkit_sse),
-    field("peers", &Health::peers), field("warning_schema_version", &Health::warning_schema_version),
-    field("warnings", &Health::warnings));
+    field("peers", &Health::peers),
+    field("x-medkit-instance-id", &Health::x_medkit_instance_id,
+          "Random id of this gateway process, present when aggregation is enabled. Generated at startup and "
+          "carried on X-Medkit-Fan-Out-Visited by the global fault list, so a gateway already asked is not asked "
+          "again."),
+    field("warning_schema_version", &Health::warning_schema_version), field("warnings", &Health::warnings));
 
 template <>
 inline constexpr std::string_view dto_name<Health> = "HealthStatus";