fix(dc_bridge): carry Measurement custom keys onto File metadata Records

A Measurement's `custom_key_str_list` keys reached its Records but not the
Uploader's `file_status`/`group_complete` rows, which were built from a fixed
field set — so the Records and the Files of one Measurement ended up labelled
differently, with nothing logged.

A Record now names its custom keys in a `custom_keys` field, which is the only
way the Uploader can tell `site` from a measured field. Those keys are appended
to every File metadata row, including retention's shed rows.

Dropped rather than written, in both cases from every row kind so a File and its
group marker can't disagree:

- a key naming a field the Uploader computes itself (`storage_type`, `size`, …)
  — the Uploader's value is kept and the Bridge warns;
- `name`, `id` and `robot_name`, which the rows already carry as `group_name`,
  `robot_id` and `robot_name` — re-emitting `id` would put a robot identifier in
  whatever an `id` column happens to be.

A Measurement with no custom keys produces the rows it did before.

Closes #419

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015oGKFjBD4WKxMHYv3iKXBo
Signed-off-by: David Bensoussan <d.bensoussan@proton.me>

Author: Minipada

dc_bridge/include/dc_bridge/uploader/group.hpp

@@ -36,6 +36,10 @@ struct FileGroup
   std::string group_name;
   std::optional<std::string> robot_name;
   std::optional<nlohmann::json> robot_id;
+  /// The emitting Measurement's `custom_key_str_list` keys and their values, read from the
+  /// Record's `custom_keys` declaration (#419). The Uploader carries them onto its File
+  /// metadata Records so a Measurement's Files and its Records are labelled the same way.
+  std::map<std::string, nlohmann::json> custom_keys;
   std::vector<FileRef> files;  ///< deterministic (local-path-sorted) order.
 };
 

dc_bridge/include/dc_bridge/uploader/status.hpp

@@ -52,6 +52,15 @@ nlohmann::json shed_row(const FileGroup& group, const FileRef& file, const Stora
 /// ADR-0005's group completion marker.
 nlohmann::json group_complete_row(const FileGroup& group);
 
+/// True for a field name the rows above compute themselves. A Measurement custom key
+/// (#419) with such a name is dropped rather than overwriting the Uploader's own value —
+/// and dropped from every row kind, not just the ones that happen to carry that field, so
+/// a File and the group marker covering it never disagree about what a key means.
+/// The Record's own `name`, `id` and `robot_name` are dropped too but are not reserved:
+/// the rows already carry those values (as `group_name`, `robot_id`, `robot_name`), so
+/// there is no disagreement to report.
+bool is_reserved_field(const std::string& name);
+
 }  // namespace status
 }  // namespace dc_bridge::uploader
 

dc_bridge/include/dc_bridge/uploader/uploader.hpp

@@ -63,6 +63,10 @@ struct ProcessSummary
   /// log line rather than by diffing the object store.
   std::size_t thumbnails = 0;
   std::size_t thumbnails_failed = 0;
+  /// Custom keys (#419) the rows could not carry because they name a field the Uploader
+  /// emits itself. Reported so the collision is visible in the Bridge's log instead of
+  /// being the silent drop this whole feature exists to remove.
+  std::vector<std::string> dropped_custom_keys;
 };
 
 /// Thrown by process_record when a Record couldn't be fully processed. `Incomplete` is

dc_bridge/src/bridge_node.cpp

@@ -682,6 +682,20 @@ void BridgeNode::run_uploader_worker(std::string forward_host, std::uint16_t for
                     item->tag.c_str(), summary.files, summary.verified, summary.missing, summary.deleted,
                     static_cast<int>(summary.group_complete));
       }
+      // A custom key naming a field the Uploader emits itself keeps the Uploader's value
+      // (#419); saying so here is the difference between a defined rule and a silent drop.
+      if (!summary.dropped_custom_keys.empty())
+      {
+        std::string joined;
+        for (const auto& key : summary.dropped_custom_keys)
+        {
+          joined += (joined.empty() ? "" : ", ") + key;
+        }
+        RCLCPP_WARN(this->get_logger(),
+                    "uploader: group '%s': custom key(s) %s name fields the File metadata Records already carry — "
+                    "the Uploader's own values are kept and the custom values are not written",
+                    item->tag.c_str(), joined.c_str());
+      }
       // Previews are best-effort and never fail an upload (#256), so a File the operator
       // asked to have one and didn't get is only visible here. Warn rather than info:
       // silently shipping galleries with no previews is the failure mode worth surfacing.

dc_bridge/src/uploader/group.cpp

@@ -120,6 +120,26 @@ FileGroup parse_file_group(const nlohmann::json& payload, const std::string& fal
     {
       group.robot_id = *id_it;
     }
+    // Top level only: a Group-merged Record namespaces its members' fields under their
+    // `group_key`, so those keys are no longer the Measurement's own labelling.
+    auto declared_it = payload.find("custom_keys");
+    if (declared_it != payload.end() && declared_it->is_array())
+    {
+      for (const auto& name : *declared_it)
+      {
+        if (!name.is_string())
+        {
+          continue;
+        }
+        const std::string key = name.get<std::string>();
+        auto value_it = payload.find(key);
+        if (key.empty() || value_it == payload.end())
+        {
+          continue;
+        }
+        group.custom_keys.emplace(key, *value_it);
+      }
+    }
   }
   for (auto& [local_path, file] : files)
   {

dc_bridge/src/uploader/status.cpp

@@ -4,13 +4,48 @@
 #include "dc_bridge/uploader/status.hpp"
 
 #include <chrono>
+#include <set>
 
 namespace dc_bridge::uploader::status
 {
 
 namespace
 {
 
+// Fields the rows compute themselves. A custom key naming one of these means the
+// Measurement and the Uploader disagree about what the name stands for.
+const std::set<std::string>& computed_fields()
+{
+  static const std::set<std::string> fields{
+    "kind",       "group_name", "robot_id", "local_path",    "remote_path",    "storage_type",
+    "updated_at", "uploaded",   "deleted",  "on_filesystem", "duration",       "content_type",
+    "size",       "complete",   "files",    "file_count",    "thumbnail_path",
+  };
+  return fields;
+}
+
+// Record keys the rows already carry, verbatim or under a Humble-era name of their own
+// (`name` → group_name, `id` → robot_id). Also dropped, but not a collision: the value is
+// in the row already, so re-emitting it would only put it in a second column.
+const std::set<std::string>& payload_derived_fields()
+{
+  static const std::set<std::string> keys{ "name", "id", "robot_name" };
+  return keys;
+}
+
+// Appends the emitting Measurement's custom keys (#419). Called last, so a key that names
+// a field the row already carries is dropped instead of overwriting it.
+void apply_custom_keys(nlohmann::json& row, const FileGroup& group)
+{
+  for (const auto& [key, value] : group.custom_keys)
+  {
+    if (!computed_fields().count(key) && !payload_derived_fields().count(key))
+    {
+      row[key] = value;
+    }
+  }
+}
+
 double unix_now()
 {
   using namespace std::chrono;
@@ -61,6 +96,7 @@ nlohmann::json uploaded_row(const FileGroup& group, const FileRef& file, const S
     // same way by a consumer.
     row["thumbnail_path"] = storage.url_prefix + storage.object_key(*thumbnail_path);
   }
+  apply_custom_keys(row, group);
   return row;
 }
 
@@ -71,6 +107,7 @@ nlohmann::json missing_row(const FileGroup& group, const FileRef& file, const St
   row["uploaded"] = false;
   row["on_filesystem"] = false;
   row["deleted"] = false;
+  apply_custom_keys(row, group);
   return row;
 }
 
@@ -81,6 +118,7 @@ nlohmann::json deleted_row(const FileGroup& group, const FileRef& file, const St
   row["uploaded"] = true;
   row["on_filesystem"] = false;
   row["deleted"] = true;
+  apply_custom_keys(row, group);
   return row;
 }
 
@@ -91,6 +129,7 @@ nlohmann::json shed_row(const FileGroup& group, const FileRef& file, const Stora
   row["uploaded"] = false;
   row["on_filesystem"] = false;
   row["deleted"] = true;
+  apply_custom_keys(row, group);
   return row;
 }
 
@@ -116,7 +155,13 @@ nlohmann::json group_complete_row(const FileGroup& group)
   }
   row["files"] = std::move(files);
   row["updated_at"] = unix_now();
+  apply_custom_keys(row, group);
   return row;
 }
 
+bool is_reserved_field(const std::string& name)
+{
+  return computed_fields().count(name) > 0;
+}
+
 }  // namespace dc_bridge::uploader::status

dc_bridge/src/uploader/uploader.cpp

@@ -449,6 +449,13 @@ ProcessSummary Uploader::process_record(const nlohmann::json& payload, const std
   {
     return summary;
   }
+  for (const auto& [key, value] : group.custom_keys)
+  {
+    if (status::is_reserved_field(key))
+    {
+      summary.dropped_custom_keys.push_back(key);
+    }
+  }
 
   std::vector<std::string> failures;
   std::vector<const FileRef*> verified_files;

dc_bridge/test/uploader_test.cpp

@@ -258,6 +258,107 @@ TEST(Uploader, MetadataRecordHasHumbleShape)
   EXPECT_EQ(marker["files"][0]["local_path"], local);
 }
 
+// The payloads below carry what measurement.hpp's addCustomKeys() writes for
+// `custom_key_str_list: [site, fleet]`: the values inline, plus a `custom_keys`
+// declaration naming which of the Record's keys are labelling rather than data.
+TEST(Uploader, CustomKeysReachEveryFileMetadataRow)
+{
+  Fixture fx({ "minio" });
+  auto local = fx.write_file("img.jpg", JPEG_BYTES);
+  auto payload = camera_payload(local, { "minio" });
+  payload["site"] = "warehouse-3";
+  payload["fleet"] = "eu-west";
+  payload["custom_keys"] = json::array({ "site", "fleet" });
+
+  auto up = fx.uploader(true);  // delete_when_sent, so the deleted rows are emitted too
+  auto rows = std::make_shared<std::vector<json>>();
+  auto summary = up.process_record(payload, "dc.measurement.camera", collect_rows(rows));
+
+  EXPECT_TRUE(summary.dropped_custom_keys.empty());
+  auto file_rows = rows_of_kind(*rows, "file_status");
+  EXPECT_EQ(file_rows.size(), 2u);  // uploaded + deleted
+  for (const auto& row : file_rows)
+  {
+    EXPECT_EQ(row["site"], "warehouse-3");
+    EXPECT_EQ(row["fleet"], "eu-west");
+  }
+  auto marker = rows_of_kind(*rows, "group_complete")[0];
+  EXPECT_EQ(marker["site"], "warehouse-3");
+  EXPECT_EQ(marker["fleet"], "eu-west");
+}
+
+// `custom_key_str_list: ["robot_name", "id"]` is what every demo configures, and the rows
+// have carried both since Humble — as `robot_name` and `robot_id`. Re-emitting the Record's
+// own `id` alongside would put a robot identifier in whatever an `id` column happens to be.
+TEST(Uploader, CustomKeysAlreadyCarriedUnderTheirOwnColumnAreNotRepeated)
+{
+  Fixture fx({ "minio" });
+  auto local = fx.write_file("img.jpg", JPEG_BYTES);
+  auto payload = camera_payload(local, { "minio" });
+  payload["custom_keys"] = json::array({ "robot_name", "id" });
+
+  auto up = fx.uploader(false);
+  auto rows = std::make_shared<std::vector<json>>();
+  auto summary = up.process_record(payload, "dc.measurement.camera", collect_rows(rows));
+
+  auto row = rows_of_kind(*rows, "file_status")[0];
+  EXPECT_EQ(row["robot_id"], "r1");
+  EXPECT_EQ(row["robot_name"], "robot1");
+  EXPECT_FALSE(row.contains("id"));
+  EXPECT_FALSE(rows_of_kind(*rows, "group_complete")[0].contains("id"));
+  // Both values are in the row already, so neither is a collision to report.
+  EXPECT_TRUE(summary.dropped_custom_keys.empty());
+}
+
+TEST(Uploader, CustomKeyNamingAnUploaderFieldNeverOverwritesIt)
+{
+  Fixture fx({ "minio" });
+  auto local = fx.write_file("img.jpg", JPEG_BYTES);
+  auto payload = camera_payload(local, { "minio" });
+  payload["storage_type"] = "not-minio";
+  payload["site"] = "warehouse-3";
+  payload["custom_keys"] = json::array({ "storage_type", "site" });
+
+  auto up = fx.uploader(false);
+  auto rows = std::make_shared<std::vector<json>>();
+  auto summary = up.process_record(payload, "dc.measurement.camera", collect_rows(rows));
+
+  auto row = rows_of_kind(*rows, "file_status")[0];
+  EXPECT_EQ(row["storage_type"], "minio");  // the Uploader's own value, not the custom one
+  EXPECT_EQ(row["site"], "warehouse-3");
+  // Dropped from every row kind, including the ones that don't carry the field themselves,
+  // so a File and its group marker can't end up disagreeing about what `storage_type` means.
+  auto marker = rows_of_kind(*rows, "group_complete")[0];
+  EXPECT_FALSE(marker.contains("storage_type"));
+  EXPECT_EQ(marker["site"], "warehouse-3");
+  EXPECT_EQ(summary.dropped_custom_keys, (std::vector<std::string>{ "storage_type" }));
+}
+
+TEST(Uploader, NoCustomKeysKeepsTheRowsUnchanged)
+{
+  Fixture fx({ "minio" });
+  auto local = fx.write_file("img.jpg", JPEG_BYTES);
+  auto up = fx.uploader(false);
+  auto rows = std::make_shared<std::vector<json>>();
+  up.process_record(camera_payload(local, { "minio" }), "dc.measurement.camera", collect_rows(rows));
+
+  auto key_set = [](const json& row) {
+    std::set<std::string> keys;
+    for (auto it = row.begin(); it != row.end(); ++it)
+    {
+      keys.insert(it.key());
+    }
+    return keys;
+  };
+  EXPECT_EQ(key_set(rows_of_kind(*rows, "file_status")[0]),
+            (std::set<std::string>{ "kind", "group_name", "robot_name", "robot_id", "local_path", "remote_path",
+                                    "storage_type", "updated_at", "uploaded", "on_filesystem", "deleted",
+                                    "content_type", "size" }));
+  EXPECT_EQ(key_set(rows_of_kind(*rows, "group_complete")[0]),
+            (std::set<std::string>{ "kind", "group_name", "robot_name", "robot_id", "complete", "file_count", "files",
+                                    "updated_at" }));
+}
+
 TEST(Uploader, VerifyFailureRetriedWithoutSecondUpload)
 {
   Fixture fx({ "minio" });

dc_measurements/include/dc_measurements/measurement.hpp

@@ -511,6 +511,7 @@ class Measurement : public dc_core::Measurement
       {
         RCLCPP_ERROR_STREAM(logger_, "Error parsing JSON when adding custom keys: " << data.dump());
       }
+      json declared = json::array();
       for (auto& param : custom_keys_)
       {
         auto key = param["key"].get<std::string>();
@@ -521,7 +522,12 @@ class Measurement : public dc_core::Measurement
           json data_custom = { { key, value } };
           data.update(data_custom);
         }
+        declared.push_back(key);
       }
+      // Names the keys that are this Measurement's labelling rather than its data, so the
+      // Bridge's Uploader can carry them onto the File metadata Records too (#419) — it
+      // has no other way to tell `site` from a measured field.
+      data["custom_keys"] = std::move(declared);
       msg.data = data.dump(-1, ' ', true);
     }
   }

dc_measurements/test/test_measurement_dummy.cpp

@@ -46,13 +46,15 @@ class MeasurementDummyTest : public ::testing::Test
     boost::replace_all(data_str, "'", "\"");
     nlohmann::json data_json = nlohmann::json::parse(data_str);
     RCLCPP_INFO_STREAM(ms_node_->get_logger(), "Value: " << data_str);
+    dummy_record_ = data_json;
     dummy_message_ = data_json["message"].get<nlohmann::json::string_t>();
     dummy_callback_ = true;
   }
 
   std::shared_ptr<measurement_server::MeasurementServer> ms_node_;
   rclcpp::Subscription<dc_interfaces::msg::StringStamped>::SharedPtr sub_data_;
   std::string dummy_message_;
+  nlohmann::json dummy_record_;
 
 public:
   bool dummy_callback_{ false };
@@ -76,6 +78,44 @@ TEST_F(MeasurementDummyTest, DummyDataCorrect)
   EXPECT_EQ(dummy_message_, message);
 }
 
+// A Record with custom keys names them, so the Bridge's Uploader can carry the same
+// labelling onto the Measurement's File metadata Records (#419).
+TEST_F(MeasurementDummyTest, CustomKeysAreDeclaredInTheRecord)
+{
+  ms_node_->declare_parameter("dummy.plugin", std::string("dc_measurements/Dummy"));
+  ms_node_->declare_parameter("dummy.topic_output", std::string("/dc/measurement/dummy"));
+  ms_node_->declare_parameter("dummy.record", std::string("{\"message\": \"My message\"}"));
+  ms_node_->declare_parameter("custom_key_str_list", std::vector<std::string>{ "site" });
+  ms_node_->declare_parameter("custom_keys_str.site.name", std::string("site"));
+  ms_node_->declare_parameter("custom_keys_str.site.value", std::string("warehouse-3"));
+
+  startLifecycleNode();
+
+  while (!dummy_callback_)
+  {
+    rclcpp::spin_some(ms_node_->get_node_base_interface());
+  }
+
+  EXPECT_EQ(dummy_record_["site"], "warehouse-3");
+  EXPECT_EQ(dummy_record_["custom_keys"], nlohmann::json::array({ "site" }));
+}
+
+TEST_F(MeasurementDummyTest, NoCustomKeysLeavesTheRecordUntouched)
+{
+  ms_node_->declare_parameter("dummy.plugin", std::string("dc_measurements/Dummy"));
+  ms_node_->declare_parameter("dummy.topic_output", std::string("/dc/measurement/dummy"));
+  ms_node_->declare_parameter("dummy.record", std::string("{\"message\": \"My message\"}"));
+
+  startLifecycleNode();
+
+  while (!dummy_callback_)
+  {
+    rclcpp::spin_some(ms_node_->get_node_base_interface());
+  }
+
+  EXPECT_FALSE(dummy_record_.contains("custom_keys"));
+}
+
 TEST_F(MeasurementDummyTest, DummyDataIncorrect)
 {
   int polling_interval = 50;