fix(bridge): send Record timestamps at nanosecond resolution

The Bridge read only `msg.header.stamp.sec` and dropped `.nanosec`, then packed
it as the ingest protocol's plain integer time. Every Record's timestamp was
rounded to the second, so a Measurement polling faster than 1 Hz produced
Records that were indistinguishable in time — measured at 5 Hz: 8 Records,
3 distinct timestamps.

The wire format already supported the fix. Fluent Forward has an EventTime
extension (ext type 0x00, 4 bytes seconds + 4 nanoseconds), and the vendored
Vector 0.57.0 parses it with all nine digits intact — verified by feeding its
fluent source both frame shapes. Only the Bridge needed to change: Record now
carries seconds and nanoseconds, and the Forwarder packs EventTime.

Adds `epoch_nanos` as a new time_format, and makes it the default. `double`
cannot represent nanoseconds — a float64 has ~15-16 significant digits and
current epoch seconds spend 10 of them, so it tops out near microseconds. That
is IEEE 754, not an implementation limit. `double` and `iso8601` stay
available. Exactness also matters beyond precision: it is what makes the
timestamp usable as part of a Record identity, which #309 needs.

The schema follows: `date` becomes bigint in both init.sql files,
`to_timestamp(date)` becomes `to_timestamp(date / 1e9)` in the dashboard, and
params writing to those tables drop their explicit `time_format: "double"`.
Console-only demos keep theirs.

Regression cover: e2e_params.yaml's memory Measurement now polls at 5 Hz so
several Records land inside one second, and verify_zero_loss.py asserts a
nanosecond remainder exists, that some second holds 2+ Records (so it cannot
pass vacuously), and that no two Records share an instant. At 1 Hz a return to
whole-second stamps would not collide and would pass unnoticed.

Closes #308

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

Author: Minipada

dc_bridge/include/dc_bridge/forwarder.hpp

@@ -17,12 +17,20 @@ namespace dc_bridge
 {
 
 /// A DC Record ready to be forwarded: a shipper ingest protocol tag, a Unix timestamp
-/// (seconds since the epoch, matching the protocol's integer time), and the Record's
+/// split into whole seconds plus the nanoseconds within that second, and the Record's
 /// JSON payload (StringStamped.data, already parsed).
+///
+/// The split mirrors the ingest protocol's **EventTime** extension (4 bytes of seconds,
+/// 4 of nanoseconds), which is what the Forwarder emits. The protocol's older integer
+/// time carries whole seconds only; sending that rounded every Record's timestamp to the
+/// second, so a Measurement polling faster than 1 Hz produced Records that were
+/// indistinguishable in time (#308).
 struct Record
 {
   std::string tag;
   std::uint64_t timestamp_secs;
+  /// Nanoseconds within `timestamp_secs`; always < 1'000'000'000.
+  std::uint32_t timestamp_nanos{ 0 };
   nlohmann::json payload;
 };
 

dc_bridge/include/dc_bridge/render.hpp

@@ -36,8 +36,16 @@ inline constexpr std::uint64_t MIN_DISK_BUFFER_BYTES = 268435488ULL;
 /// route transform id, rest = the Tag verbatim).
 std::string route_output_for_tag(const std::string& tag);
 
+/// How a Destination's normalized time field is written.
+///
+/// `EpochNanos` is the default: an exact integer count of nanoseconds since the epoch.
+/// `Double` divides that by 1e9 into a float64, which has ~15-16 significant digits and
+/// spends 10 of them on the seconds — so it cannot represent better than roughly
+/// microseconds, and rounds. It stays available for consumers that want a fractional
+/// seconds column, but it is lossy by construction, not by implementation (#308).
 enum class TimeFormat
 {
+  EpochNanos,
   Double,
   Iso8601,
 };

dc_bridge/src/bridge_node.cpp

@@ -39,6 +39,19 @@ std::string expand_with_env(const std::string& input)
   return expand_env(input, env_lookup);
 }
 
+// Stamps a Bridge-generated Record (File status, retention audit) with the current time,
+// split into the seconds/nanoseconds pair the ingest protocol's EventTime carries. These
+// Records have no ROS header to take a stamp from, unlike the ones forwarded from a
+// Measurement's topic. Truncating this to whole seconds was part of #308.
+void stamp_now(Record& record)
+{
+  const auto since_epoch = std::chrono::system_clock::now().time_since_epoch();
+  const auto secs = std::chrono::duration_cast<std::chrono::seconds>(since_epoch);
+  record.timestamp_secs = static_cast<std::uint64_t>(secs.count());
+  record.timestamp_nanos =
+      static_cast<std::uint32_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(since_epoch - secs).count());
+}
+
 // Declares `name` with dynamic typing and a PARAMETER_NOT_SET default, so a value the
 // user didn't provide reads back as nullopt (rather than forcing a sentinel/default).
 std::optional<std::string> declare_optional_string(rclcpp::Node* node, const std::string& name)
@@ -419,7 +432,11 @@ BridgeNode::BridgeNode(const rclcpp::NodeOptions& options) : rclcpp::Node("dc_br
           {
             Record record;
             record.tag = tag;
+            // Both halves of the ROS stamp. The nanoseconds used to be dropped here,
+            // which rounded every Record to the second and left a Measurement polling
+            // faster than 1 Hz with Records indistinguishable in time (#308).
             record.timestamp_secs = static_cast<std::uint64_t>(std::max<std::int32_t>(0, msg.header.stamp.sec));
+            record.timestamp_nanos = msg.header.stamp.nanosec;
             record.payload = std::move(payload);
             try
             {
@@ -465,8 +482,7 @@ void BridgeNode::run_uploader_worker(std::string forward_host, std::uint16_t for
   auto emit = [&forwarder](const nlohmann::json& row) {
     Record record;
     record.tag = uploader::FILE_STATUS_TAG;
-    record.timestamp_secs = static_cast<std::uint64_t>(
-        std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count());
+    stamp_now(record);
     record.payload = row;
     // Vector may be briefly down (restart, backpressure); keep trying for a while before
     // handing the whole Record back for an idempotent retry.
@@ -493,8 +509,7 @@ void BridgeNode::run_uploader_worker(std::string forward_host, std::uint16_t for
   auto retention_emit = [&forwarder](const nlohmann::json& row) {
     Record record;
     record.tag = uploader::FILE_STATUS_TAG;
-    record.timestamp_secs = static_cast<std::uint64_t>(
-        std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count());
+    stamp_now(record);
     record.payload = row;
     forwarder.send(record);
   };

dc_bridge/src/forwarder.cpp

@@ -9,6 +9,7 @@
 #include <sys/types.h>
 #include <unistd.h>
 
+#include <array>
 #include <cerrno>
 #include <cstring>
 #include <ctime>
@@ -87,6 +88,29 @@ void pack_record_map(Packer& pk, const nlohmann::json& payload)
   }
 }
 
+// Packs a timestamp as the ingest protocol's **EventTime** extension: msgpack ext type
+// 0x00 with an 8-byte body — 4 bytes of seconds then 4 of nanoseconds, both big-endian
+// (network order), exactly as the protocol specifies.
+//
+// The alternative the protocol allows is a plain integer of whole seconds, which is what
+// DC sent until #308. That silently rounded every Record's timestamp down to the second,
+// so five Records from a 5 Hz Measurement all claimed the same instant. Seconds are
+// narrowed to 32 bits here because the extension's layout is fixed at 4 bytes; that field
+// overflows in 2106, long after the int32 ROS header stamp this is built from.
+template <typename Packer>
+void pack_event_time(Packer& pk, std::uint64_t seconds, std::uint32_t nanos)
+{
+  std::array<char, 8> body{};
+  const auto secs32 = static_cast<std::uint32_t>(seconds);
+  for (int i = 0; i < 4; ++i)
+  {
+    body[i] = static_cast<char>((secs32 >> (8 * (3 - i))) & 0xFF);
+    body[4 + i] = static_cast<char>((nanos >> (8 * (3 - i))) & 0xFF);
+  }
+  pk.pack_ext(body.size(), 0x00);
+  pk.pack_ext_body(body.data(), body.size());
+}
+
 }  // namespace
 
 Forwarder::Forwarder(ForwarderConfig config) : config_(std::move(config))
@@ -153,7 +177,7 @@ std::string Forwarder::frame(const Record& record, const std::string& chunk_id)
   pk.pack(record.tag);
   pk.pack_array(1);  // one entry
   pk.pack_array(2);  // [time, record]
-  pk.pack(record.timestamp_secs);
+  pack_event_time(pk, record.timestamp_secs, record.timestamp_nanos);
   pack_record_map(pk, record.payload);
   pk.pack_map(1);
   pk.pack(std::string("chunk"));

dc_bridge/src/render.cpp

@@ -134,7 +134,14 @@ std::string render_normalize_source(const RenderConfig& config)
   for (const auto& dest : config.destinations)
   {
     out += "if " + tag_condition(destination_tags(dest)) + " {\n";
-    if (dest.time_format == TimeFormat::Double)
+    if (dest.time_format == TimeFormat::EpochNanos)
+    {
+      // Exact: an i64 of nanoseconds since the epoch, no float rounding anywhere. Good
+      // until the year 2262, and the only format here that can round-trip the full
+      // resolution the Forwarder now sends (#308).
+      out += "  ." + dest.time_key + " = to_unix_timestamp!(.timestamp, unit: \"nanoseconds\")\n";
+    }
+    else if (dest.time_format == TimeFormat::Double)
     {
       out +=
           "  ." + dest.time_key + " = to_float(to_unix_timestamp!(.timestamp, unit: \"nanoseconds\")) / 1000000000.0\n";
@@ -430,9 +437,13 @@ Destination destination_from_raw(const std::string& name, const std::string& typ
   }
 
   std::string time_key = optional_field(raw.time_key).value_or("date");
-  std::string tf = raw.time_format && !raw.time_format->empty() ? *raw.time_format : "double";
+  std::string tf = raw.time_format && !raw.time_format->empty() ? *raw.time_format : "epoch_nanos";
   TimeFormat time_format;
-  if (tf == "double")
+  if (tf == "epoch_nanos")
+  {
+    time_format = TimeFormat::EpochNanos;
+  }
+  else if (tf == "double")
   {
     time_format = TimeFormat::Double;
   }
@@ -442,9 +453,10 @@ Destination destination_from_raw(const std::string& name, const std::string& typ
   }
   else
   {
-    throw RenderError(
-        RenderErrorKind::InvalidTimeFormat,
-        "destination '" + name + "': time_format '" + tf + "' is invalid (expected 'double' or 'iso8601')", name, tf);
+    throw RenderError(RenderErrorKind::InvalidTimeFormat,
+                      "destination '" + name + "': time_format '" + tf +
+                          "' is invalid (expected 'epoch_nanos', 'double' or 'iso8601')",
+                      name, tf);
   }
 
   DestinationKind kind;

dc_bridge/test/forwarder_test.cpp

@@ -25,9 +25,29 @@ using namespace dc_bridge;
 namespace
 {
 
-Record make_record(const std::string& tag, const std::string& json)
+Record make_record(const std::string& tag, const std::string& json, std::uint32_t nanos = 0)
 {
-  return Record{ tag, 1700000000ULL, nlohmann::json::parse(json) };
+  return Record{ tag, 1700000000ULL, nanos, nlohmann::json::parse(json) };
+}
+
+// The EventTime extension the frame carries the timestamp in: msgpack ext type 0x00,
+// 8-byte body of big-endian seconds then big-endian nanoseconds (#308). Built here
+// independently of the packing code so the test pins the wire bytes rather than
+// re-deriving them the same way the implementation does.
+std::string expected_event_time_bytes(std::uint32_t secs, std::uint32_t nanos)
+{
+  std::string out;
+  out.push_back(static_cast<char>(0xd7));  // fixext8
+  out.push_back(static_cast<char>(0x00));  // ext type 0 = EventTime
+  for (int i = 0; i < 4; ++i)
+  {
+    out.push_back(static_cast<char>((secs >> (8 * (3 - i))) & 0xFF));
+  }
+  for (int i = 0; i < 4; ++i)
+  {
+    out.push_back(static_cast<char>((nanos >> (8 * (3 - i))) & 0xFF));
+  }
+  return out;
 }
 
 // A mock shipper ingest protocol server: binds :0, hands back the chosen port, and runs
@@ -147,7 +167,8 @@ TEST(Forwarder, FrameIsWellFormedWithCorrectTag)
   msgpack::object entries = top.via.array.ptr[1];
   ASSERT_EQ(entries.via.array.size, 1u);
   msgpack::object entry = entries.via.array.ptr[0];  // [time, record]
-  EXPECT_EQ(entry.via.array.ptr[0].as<std::uint64_t>(), 1700000000ULL);
+  // The time element is the EventTime extension, not a bare integer of seconds (#308).
+  EXPECT_EQ(entry.via.array.ptr[0].type, msgpack::type::EXT);
 
   msgpack::object rec = entry.via.array.ptr[1];
   ASSERT_EQ(rec.type, msgpack::type::MAP);
@@ -163,6 +184,36 @@ TEST(Forwarder, FrameIsWellFormedWithCorrectTag)
   EXPECT_TRUE(found);
 }
 
+// #308: sub-second resolution must survive the wire. Before this, the frame carried a
+// plain integer of whole seconds, so a Measurement polling faster than 1 Hz produced
+// Records that all claimed the same instant.
+TEST(Forwarder, FrameCarriesSubSecondPrecisionAsEventTime)
+{
+  const std::uint32_t nanos = 123456789u;
+  const std::string frame = Forwarder::frame(make_record("dc.test", R"({"n": 1})", nanos), "chunk-ns");
+  EXPECT_NE(frame.find(expected_event_time_bytes(1700000000u, nanos)), std::string::npos)
+      << "frame does not carry the EventTime extension for 1700000000.123456789";
+}
+
+// Two Records one nanosecond apart must not collapse onto the same wire timestamp —
+// the property that makes a timestamp usable as part of an identity.
+TEST(Forwarder, FramesWithinTheSameSecondAreDistinguishable)
+{
+  const std::string a = Forwarder::frame(make_record("dc.test", R"({"n": 1})", 1u), "c1");
+  const std::string b = Forwarder::frame(make_record("dc.test", R"({"n": 1})", 2u), "c1");
+  EXPECT_NE(a, b);
+  EXPECT_NE(a.find(expected_event_time_bytes(1700000000u, 1u)), std::string::npos);
+  EXPECT_NE(b.find(expected_event_time_bytes(1700000000u, 2u)), std::string::npos);
+}
+
+// A whole-second Record still encodes as EventTime, with a zero nanosecond half, rather
+// than falling back to the integer form.
+TEST(Forwarder, WholeSecondTimestampStillUsesEventTime)
+{
+  const std::string frame = Forwarder::frame(make_record("dc.test", R"({"n": 1})", 0u), "c0");
+  EXPECT_NE(frame.find(expected_event_time_bytes(1700000000u, 0u)), std::string::npos);
+}
+
 TEST(Forwarder, FrameCarriesTheChunkOption)
 {
   auto oh = unpack(Forwarder::frame(make_record("dc.test", R"({"n": 1})"), "my-chunk-id"));
@@ -258,7 +309,7 @@ TEST(Forwarder, ReturnsBackpressureWhenPeerStalls)
 
   nlohmann::json big;
   big["blob"] = std::string(1000000, 'x');
-  Record big_record{ "dc.test", 1700000000ULL, big };
+  Record big_record{ "dc.test", 1700000000ULL, 0u, big };
 
   bool saw_backpressure = false;
   for (int i = 0; i < 20; ++i)

dc_bridge/test/render_test.cpp

@@ -417,6 +417,33 @@ TEST(Render, DestinationFromRawRejectsInvalidTimeFormat)
   }
 }
 
+TEST(Render, DestinationFromRawAcceptsEveryTimeFormat)
+{
+  const std::vector<std::pair<std::string, TimeFormat>> cases = { { "epoch_nanos", TimeFormat::EpochNanos },
+                                                                  { "double", TimeFormat::Double },
+                                                                  { "iso8601", TimeFormat::Iso8601 } };
+  for (const auto& c : cases)
+  {
+    RawDestinationParams raw;
+    raw.time_format = c.first;
+    auto dest = destination_from_raw("debug_console", "console", "records", { "/dc/measurement/map" }, raw);
+    EXPECT_EQ(dest.time_format, c.second) << "time_format '" << c.first << "'";
+  }
+}
+
+// The normalize transform must emit the timestamp with no float arithmetic in it —
+// `to_float(...) / 1000000000.0` is precisely what loses resolution below ~1 us.
+TEST(Render, EpochNanosNormalizesWithoutFloatRounding)
+{
+  auto config =
+      config_with({ make_destination("pgsql", { "/dc/group/robot" }, TimeFormat::EpochNanos, postgres_kind()) });
+  const std::string rendered = render(config);
+  EXPECT_NE(rendered.find(".date = to_unix_timestamp!(.timestamp, unit: \"nanoseconds\")"), std::string::npos)
+      << rendered;
+  EXPECT_EQ(rendered.find("to_float("), std::string::npos)
+      << "epoch_nanos must not round through a float: " << rendered;
+}
+
 TEST(Render, PostgresFromRawRejectsMissingRequiredFields)
 {
   RawDestinationParams raw;
@@ -465,11 +492,13 @@ TEST(Render, PostgresFromRawAppliesDefaults)
   EXPECT_EQ(pg.port, 5432);
 }
 
+// #308: the default is the exact integer format, not the lossy float one. A Destination
+// that says nothing about time must not silently round its timestamps.
 TEST(Render, DestinationFromRawAppliesTimeDefaults)
 {
   auto dest = destination_from_raw("debug_console", "console", "records", { "/dc/group/robot" }, {});
   EXPECT_EQ(dest.time_key, "date");
-  EXPECT_EQ(dest.time_format, TimeFormat::Double);
+  EXPECT_EQ(dest.time_format, TimeFormat::EpochNanos);
 }
 
 TEST(Render, S3FromRawRejectsMissingBucket)

dc_bringup/params/dc_params.yaml

@@ -25,7 +25,10 @@ dc_bridge:
       database: "dc"
       table: "dc"
       time_key: "date"
-      time_format: "double"
+      # epoch_nanos (default) = exact integer nanoseconds since the epoch; iso8601 =
+      # "…T11:36:28.123456789" string; double = fractional seconds, which is a float64
+      # and rounds below ~1 us. Match the column type: bigint for epoch_nanos.
+      time_format: "epoch_nanos"
     # The other blessed destination types (add the name to `destinations` to enable):
     #
     # rustfs:                     # S3-compatible object storage (self-hosted: RustFS

dc_demos/params/qrcodes_minio_pgsql.yaml

@@ -20,7 +20,6 @@ dc_bridge:
       database: "dc"
       table: "dc"
       time_key: "date"
-      time_format: "double"
     pgsql_files:
       type: postgres
       receives: records
@@ -31,7 +30,6 @@ dc_bridge:
       database: "dc"
       table: "dc_files"
       time_key: "date"
-      time_format: "double"
     rustfs:
       type: s3
       receives: files

dc_demos/params/tb3_simulation_pgsql_minio.yaml

@@ -31,7 +31,6 @@ dc_bridge:
       database: "dc"
       table: "dc"
       time_key: "date"
-      time_format: "double"
     pgsql_files:
       type: postgres
       receives: records
@@ -42,7 +41,6 @@ dc_bridge:
       database: "dc"
       table: "dc_files"
       time_key: "date"
-      time_format: "double"
     rustfs:
       type: s3
       receives: files