refactor(dc_measurements): deepen the Mission adapter lifecycle and pending-record queue

dc_measurements/include/dc_measurements/measurement.hpp

@@ -50,6 +50,30 @@ enum class Status : int8_t
   RUNNING = 3,
 };
 
+// A json_validator schema_loader resolving a "$ref" naming a sibling file (e.g.
+// "mission_base.json") against `schema_dir` -- the directory the root schema itself was loaded
+// from. Every schema in this codebase but the Mission Measurement family keeps to same-file
+// "#/$defs/..." refs, which nlohmann_json_schema_validator resolves on its own; this loader is
+// only ever invoked for a schema that references another file, letting a family of adapters
+// (#305/ADR-0010's shared mission_start/mission_end contract) share one base schema instead of
+// duplicating its properties per adapter.
+inline nlohmann::json_schema::schema_loader makeSchemaFileLoader(const std::string& schema_dir)
+{
+  return [schema_dir](const nlohmann::json_uri& id, json& value) {
+    std::string filename = id.path();
+    if (!filename.empty() && filename.front() == '/')
+    {
+      filename.erase(0, 1);
+    }
+    std::ifstream f(schema_dir + "/" + filename);
+    if (!f)
+    {
+      throw std::runtime_error{ "could not load schema '" + filename + "' referenced from " + schema_dir };
+    }
+    value = json::parse(f);
+  };
+}
+
 /**
  * @class nav2_behaviors::Behavior
  * @brief An action server Behavior base class implementing the action server and basic factory.
@@ -144,7 +168,8 @@ class Measurement : public dc_core::Measurement
   void validateSchema(const std::string& package_name, const std::string& json_filename)
   {
     std::string package_share_directory = ament_index_cpp::get_package_share_directory(package_name);
-    std::string path = package_share_directory + "/plugins/measurements/json/" + json_filename;
+    std::string schema_dir = package_share_directory + "/plugins/measurements/json";
+    std::string path = schema_dir + "/" + json_filename;
     std::ifstream f(path.c_str());
     try
     {
@@ -155,6 +180,7 @@ class Measurement : public dc_core::Measurement
       RCLCPP_INFO_STREAM(logger_, "schema: " << schema_);
       try
       {
+        validator_ = json_validator(makeSchemaFileLoader(schema_dir));
         validator_.set_root_schema(schema_);
       }
       catch (const std::exception& e)
@@ -180,6 +206,8 @@ class Measurement : public dc_core::Measurement
       RCLCPP_INFO_STREAM(logger_, "schema: " << schema_);
       try
       {
+        const std::string schema_dir = std::filesystem::path(json_schema_path).parent_path().string();
+        validator_ = json_validator(makeSchemaFileLoader(schema_dir));
         validator_.set_root_schema(schema_);
       }
       catch (const std::exception& e)

dc_measurements/include/dc_measurements/mission_outcome.hpp

@@ -0,0 +1,53 @@
+// SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+// SPDX-License-Identifier: MPL-2.0
+
+#ifndef DC_MEASUREMENTS__MISSION_OUTCOME_HPP_
+#define DC_MEASUREMENTS__MISSION_OUTCOME_HPP_
+
+#include <cstdint>
+#include <string>
+
+namespace dc_measurements
+{
+
+/// The outcome a Mission Record reports on mission_end, per the mission lifecycle contract
+/// (#305, recorded in ADR-0010). Shared by every Mission Measurement adapter (nav2's three
+/// action variants, Open-RMF) so the vocabulary -- and what a caller can do with it -- stays one
+/// type rather than four independently-declared look-alikes.
+enum class MissionOutcome
+{
+  Succeeded,
+  Failed,
+  Cancelled,
+  Aborted,
+};
+
+/// The JSON-Record spelling of a MissionOutcome, per #305/ADR-0010's Record shape.
+inline std::string missionOutcomeName(MissionOutcome outcome)
+{
+  switch (outcome)
+  {
+    case MissionOutcome::Succeeded:
+      return "succeeded";
+    case MissionOutcome::Failed:
+      return "failed";
+    case MissionOutcome::Cancelled:
+      return "cancelled";
+    case MissionOutcome::Aborted:
+      return "aborted";
+  }
+  return "unknown";
+}
+
+/// A mission_start fact: identical across every adapter (nav2's three action variants,
+/// Open-RMF) -- only what varies per source (the terminal outcome, its reason, extra fields
+/// like recoveries/missed_waypoints) lives in each adapter's own MissionEndFact.
+struct MissionStartFact
+{
+  std::string mission_id;
+  std::uint64_t sequence{ 0 };
+};
+
+}  // namespace dc_measurements
+
+#endif  // DC_MEASUREMENTS__MISSION_OUTCOME_HPP_

dc_measurements/include/dc_measurements/mission_record_json.hpp

@@ -0,0 +1,58 @@
+// SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+// SPDX-License-Identifier: MPL-2.0
+
+#ifndef DC_MEASUREMENTS__MISSION_RECORD_JSON_HPP_
+#define DC_MEASUREMENTS__MISSION_RECORD_JSON_HPP_
+
+#include <cstdint>
+#include <nlohmann/json.hpp>
+#include <optional>
+#include <string>
+
+#include "dc_measurements/mission_outcome.hpp"
+
+namespace dc_measurements
+{
+
+/// The mission_start Record every Mission Measurement adapter emits, per #305/ADR-0010's shared
+/// Record contract.
+inline nlohmann::json missionStartJson(const std::string& mission_id, const std::string& mission_type,
+                                       std::uint64_t sequence)
+{
+  nlohmann::json data;
+  data["event"] = "mission_start";
+  data["mission_id"] = mission_id;
+  data["mission_type"] = mission_type;
+  data["sequence"] = sequence;
+  return data;
+}
+
+/// The mission_end Record fields common to every adapter (#305/ADR-0010). A caller with its own
+/// extra fields (nav2's `recoveries`, FollowWaypoints' `missed_waypoints`) adds them to the
+/// returned object before enqueuing it.
+inline nlohmann::json missionEndJsonBase(const std::string& mission_id, const std::string& mission_type,
+                                         std::uint64_t sequence, MissionOutcome outcome, double duration_sec,
+                                         const std::optional<std::string>& reason,
+                                         const std::optional<std::uint16_t>& error_code)
+{
+  nlohmann::json data;
+  data["event"] = "mission_end";
+  data["mission_id"] = mission_id;
+  data["mission_type"] = mission_type;
+  data["sequence"] = sequence;
+  data["outcome"] = missionOutcomeName(outcome);
+  data["duration_sec"] = duration_sec;
+  if (reason.has_value())
+  {
+    data["reason"] = *reason;
+  }
+  if (error_code.has_value())
+  {
+    data["error_code"] = *error_code;
+  }
+  return data;
+}
+
+}  // namespace dc_measurements
+
+#endif  // DC_MEASUREMENTS__MISSION_RECORD_JSON_HPP_

dc_measurements/include/dc_measurements/mission_registry.hpp

@@ -0,0 +1,103 @@
+// SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+// SPDX-License-Identifier: MPL-2.0
+
+#ifndef DC_MEASUREMENTS__MISSION_REGISTRY_HPP_
+#define DC_MEASUREMENTS__MISSION_REGISTRY_HPP_
+
+#include <cstddef>
+#include <cstdint>
+#include <deque>
+#include <map>
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace dc_measurements
+{
+
+/// Bookkeeping shared by every Mission Measurement core that can track more than one mission at
+/// once (MissionNav2ThroughPosesCore, MissionOpenRmfCore): a bounded id -> ActivePayload map,
+/// oldest-finished-first eviction once `max_tracked` ids have ever been seen, and the monotonic
+/// sequence counter every mission_start/mission_end fact carries. `ActivePayload` is whatever
+/// per-mission state one core needs beyond "when did it start and is it finished" -- it must
+/// expose a `bool finished` member; PrunedMissionMap never interprets any other field.
+///
+/// Owns no clock and no ROS/json type, matching every other tracker/core in this codebase: a
+/// unit test drives it in microseconds with no node or action server involved.
+template <typename ActivePayload>
+class PrunedMissionMap
+{
+public:
+  explicit PrunedMissionMap(std::size_t max_tracked = 256) : max_tracked_(max_tracked)
+  {
+  }
+
+  ActivePayload* find(const std::string& id)
+  {
+    auto it = missions_.find(id);
+    return it == missions_.end() ? nullptr : &it->second;
+  }
+
+  /// Starts tracking `id`. Caller must already have checked find(id) == nullptr -- mirrors every
+  /// existing core's own observe()/end() call pattern, so this never silently overwrites an
+  /// in-flight mission's state.
+  ActivePayload& insert(const std::string& id, ActivePayload payload)
+  {
+    auto [it, inserted] = missions_.emplace(id, std::move(payload));
+    order_.push_back(id);
+    prune();
+    return it->second;
+  }
+
+  std::uint64_t nextSequence()
+  {
+    return ++sequence_;
+  }
+
+  /// ids whose payload is not yet finished -- for onCleanup()'s "still running at shutdown"
+  /// warning, the same shape every Mission core already reports it in.
+  std::vector<std::string> openIds() const
+  {
+    std::vector<std::string> open;
+    for (const auto& [id, payload] : missions_)
+    {
+      if (!payload.finished)
+      {
+        open.push_back(id);
+      }
+    }
+    return open;
+  }
+
+private:
+  // ids never repeat in practice (UUIDs / Open-RMF booking ids), so a long-running instance's map
+  // would otherwise grow without bound; only ever prunes finished missions, oldest first.
+  void prune()
+  {
+    while (order_.size() > max_tracked_)
+    {
+      const auto& oldest = order_.front();
+      auto it = missions_.find(oldest);
+      if (it != missions_.end() && it->second.finished)
+      {
+        missions_.erase(it);
+        order_.pop_front();
+      }
+      else
+      {
+        // The oldest tracked id is still open (a very long-running mission) -- leave it rather
+        // than dropping an active mission's tracking state.
+        break;
+      }
+    }
+  }
+
+  std::size_t max_tracked_;
+  std::map<std::string, ActivePayload> missions_;
+  std::deque<std::string> order_;
+  std::uint64_t sequence_{ 0 };
+};
+
+}  // namespace dc_measurements
+
+#endif  // DC_MEASUREMENTS__MISSION_REGISTRY_HPP_

dc_measurements/include/dc_measurements/mission_uuid.hpp

@@ -0,0 +1,53 @@
+// SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+// SPDX-License-Identifier: MPL-2.0
+
+#ifndef DC_MEASUREMENTS__MISSION_UUID_HPP_
+#define DC_MEASUREMENTS__MISSION_UUID_HPP_
+
+#include <array>
+#include <cstdint>
+#include <iomanip>
+#include <sstream>
+#include <string>
+
+#include "unique_identifier_msgs/msg/uuid.hpp"
+
+namespace dc_measurements
+{
+
+/// A goal UUID as RFC-4122-style dashed lowercase hex (`xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`).
+/// Used as `mission_id` by the single-goal-at-a-time nav2 adapters (NavigateToPose,
+/// FollowWaypoints), which format directly from the action's own UUID message.
+inline std::string missionGoalIdDashed(const unique_identifier_msgs::msg::UUID& uuid)
+{
+  std::ostringstream oss;
+  oss << std::hex << std::setfill('0');
+  for (std::size_t i = 0; i < uuid.uuid.size(); ++i)
+  {
+    oss << std::setw(2) << static_cast<int>(uuid.uuid[i]);
+    if (i == 3 || i == 5 || i == 7 || i == 9)
+    {
+      oss << '-';
+    }
+  }
+  return oss.str();
+}
+
+/// A goal UUID as plain (non-dashed) lowercase hex. Used as `mission_id` by
+/// MissionNav2ThroughPoses, which formats from `goal_info.goal_id.uuid` directly rather than the
+/// full UUID message type. Kept distinct from missionGoalIdDashed() rather than unified: an
+/// already-shipped adapter's mission_id format is a Record-shape detail downstream consumers may
+/// already depend on, not an implementation detail free to change underneath them.
+inline std::string missionGoalIdHex(const std::array<std::uint8_t, 16>& uuid)
+{
+  std::ostringstream oss;
+  for (auto byte : uuid)
+  {
+    oss << std::hex << std::setfill('0') << std::setw(2) << static_cast<int>(byte);
+  }
+  return oss.str();
+}
+
+}  // namespace dc_measurements
+
+#endif  // DC_MEASUREMENTS__MISSION_UUID_HPP_

dc_measurements/include/dc_measurements/pending_record_queue.hpp

@@ -0,0 +1,64 @@
+// SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+// SPDX-License-Identifier: MPL-2.0
+
+#ifndef DC_MEASUREMENTS__PENDING_RECORD_QUEUE_HPP_
+#define DC_MEASUREMENTS__PENDING_RECORD_QUEUE_HPP_
+
+#include <cstddef>
+#include <deque>
+#include <nlohmann/json.hpp>
+#include <utility>
+
+#include "rclcpp/time.hpp"
+
+namespace dc_measurements
+{
+
+/// A bounded FIFO of collected-but-not-yet-published Records, drained one per poll via pop() --
+/// the shape every Measurement that can emit more than one Record between polls (Fault,
+/// Ros2ControlStatus, the Mission adapters) already needed, previously reimplemented per plugin.
+/// One Record leaves per poll, so a source producing Records far faster than the polling interval
+/// would otherwise queue without bound: past `max_size`, push() drops the oldest entry -- the
+/// most recent boundaries are the ones still worth reporting -- and reports that back to the
+/// caller, which owns the actual logger/measurement-name-specific warning text.
+class PendingRecordQueue
+{
+public:
+  explicit PendingRecordQueue(std::size_t max_size = 64) : max_size_(max_size)
+  {
+  }
+
+  /// Queues `data` for emission at `stamp`. Returns true when this push dropped the oldest queued
+  /// entry to stay within `max_size`, so the caller can log its own throttled warning.
+  bool push(nlohmann::json data, const rclcpp::Time& stamp)
+  {
+    records_.emplace_back(std::move(data), stamp);
+    if (records_.size() > max_size_)
+    {
+      records_.pop_front();
+      return true;
+    }
+    return false;
+  }
+
+  bool empty() const
+  {
+    return records_.empty();
+  }
+
+  /// Pops the oldest queued entry. Caller must check empty() first.
+  std::pair<nlohmann::json, rclcpp::Time> pop()
+  {
+    auto record = std::move(records_.front());
+    records_.pop_front();
+    return record;
+  }
+
+private:
+  std::size_t max_size_;
+  std::deque<std::pair<nlohmann::json, rclcpp::Time>> records_;
+};
+
+}  // namespace dc_measurements
+
+#endif  // DC_MEASUREMENTS__PENDING_RECORD_QUEUE_HPP_