FIX: Fix ts_out decoding with DBNv1 upgrade

Author: threecgreen

CHANGELOG.md

@@ -4,13 +4,18 @@
 
 ### Enhancements
 - Added new publisher values for consolidated DBEQ.MAX
+- Added constructor to `WithTsOut` that updates `length` to the correct value to account
+  for the extra 8 bytes
 
 ### Breaking changes
 - Changed default `upgrade_policy` to `Upgrade` so by default the primary record types
   can always be used
 - Renamed `dummy` field in `ImbalanceMsg` and `StatMsg` to `reserved`
 
 ### Bug fixes
+- Fixed handling of `ts_out` when decoding DBNv1 and upgrading to version 2
+- Fixed missing logic to upgrade `ErrorMsgV1` and `SystemMsgV1` when decoding DBN with
+  `VersionUpgradePolicy::Upgrade`
 - Added missing `StatType::Vwap` variant used in the ICE datasets
 - Added missing `ToString` and `operator<<` handling for `StatType::ClosePrice` and
   `StatType::NetChange`

include/databento/constants.hpp

@@ -24,6 +24,8 @@ static constexpr auto kUndefTimestamp =
 static constexpr auto kDbnVersion = 2;
 // The length of fixed-length symbol strings.
 static constexpr auto kSymbolCstrLen = 71;
+// The multiplier for converting the `length` field in `RecordHeader` to bytes.
+static constexpr std::size_t kRecordHeaderLengthMultiplier = 4;
 
 // This is not necessarily a comprehensive list of available datasets. Please
 // use `Historical.MetadataListDatasets` to retrieve an up-to-date list.

include/databento/dbn_decoder.hpp

@@ -33,7 +33,7 @@ class DbnDecoder {
   // given version and upgrade policy. If an upgrade is applied,
   // compat_buffer is modified.
   static Record DecodeRecordCompat(
-      std::uint8_t version, VersionUpgradePolicy upgrade_policy,
+      std::uint8_t version, VersionUpgradePolicy upgrade_policy, bool ts_out,
       std::array<std::uint8_t, kMaxRecordLen>* compat_buffer, Record rec);
 
   // Should be called exactly once.
@@ -64,6 +64,7 @@ class DbnDecoder {
 
   std::uint8_t version_{};
   VersionUpgradePolicy upgrade_policy_;
+  bool ts_out_{};
   std::unique_ptr<IReadable> input_;
   std::vector<std::uint8_t> read_buffer_;
   std::size_t buffer_idx_{};

include/databento/record.hpp

@@ -13,11 +13,13 @@
 #include "databento/enums.hpp"
 #include "databento/flag_set.hpp"    // FlagSet
 #include "databento/publishers.hpp"  // Publisher
+#include "databento/with_ts_out.hpp"
 
 namespace databento {
 // Common data for all Databento Records.
 struct RecordHeader {
-  static constexpr std::size_t kLengthMultiplier = 4;
+  static constexpr std::size_t kLengthMultiplier =
+      kRecordHeaderLengthMultiplier;
 
   // The length of the message in 32-bit words.
   std::uint8_t length;
@@ -537,5 +539,6 @@ std::ostream& operator<<(std::ostream& stream,
                          const SymbolMappingMsg& symbol_mapping_msg);
 
 // The length in bytes of the largest record type.
-static constexpr std::size_t kMaxRecordLen = sizeof(InstrumentDefMsg);
+static constexpr std::size_t kMaxRecordLen =
+    sizeof(WithTsOut<InstrumentDefMsg>);
 }  // namespace databento

include/databento/with_ts_out.hpp

@@ -1,7 +1,8 @@
 #pragma once
 
-#include "databento/datetime.hpp"  // UnixNanos
-#include "databento/enums.hpp"
+#include "databento/constants.hpp"  // kRecordHeaderLengthMultiplier
+#include "databento/datetime.hpp"   // UnixNanos
+#include "databento/enums.hpp"      // RType
 
 namespace databento {
 // Record wrapper to read records with their live gateway send
@@ -10,6 +11,11 @@ template <typename R>
 struct WithTsOut {
   static bool HasRType(RType rtype) { return R::HasRType(rtype); }
 
+  constexpr WithTsOut(R r, UnixNanos ts) : rec{r}, ts_out{ts} {
+    // Adjust length for `ts_out`
+    this->rec.hd.length = sizeof(*this) / kRecordHeaderLengthMultiplier;
+  }
+
   // The base record.
   R rec;
   // The end timestamp from the Databento live gateway.

src/dbn_decoder.cpp

@@ -13,6 +13,7 @@
 #include "databento/enums.hpp"
 #include "databento/exceptions.hpp"
 #include "databento/record.hpp"
+#include "databento/with_ts_out.hpp"
 
 using databento::DbnDecoder;
 
@@ -188,23 +189,47 @@ databento::Metadata DbnDecoder::DecodeMetadata() {
   read_buffer_.resize(version_and_size.second);
   input_->ReadExact(read_buffer_.data(), read_buffer_.size());
   buffer_idx_ = read_buffer_.size();
-  return DbnDecoder::DecodeMetadataFields(version_, read_buffer_);
+  auto metadata = DbnDecoder::DecodeMetadataFields(version_, read_buffer_);
+  ts_out_ = metadata.ts_out;
+  return metadata;
 }
 
+namespace {
+template <typename T, typename U>
+databento::Record UpgradeRecord(
+    bool ts_out,
+    std::array<std::uint8_t, databento::kMaxRecordLen>* compat_buffer,
+    databento::Record rec) {
+  if (ts_out) {
+    const auto orig = rec.Get<databento::WithTsOut<T>>();
+    const databento::WithTsOut<U> v2{orig.rec.ToV2(), orig.ts_out};
+    const auto v2_ptr = reinterpret_cast<const std::uint8_t*>(&v2);
+    std::copy(v2_ptr, v2_ptr + v2.rec.hd.Size(), compat_buffer->data());
+  } else {
+    const auto v2 = rec.Get<T>().ToV2();
+    const auto v2_ptr = reinterpret_cast<const std::uint8_t*>(&v2);
+    std::copy(v2_ptr, v2_ptr + v2.hd.Size(), compat_buffer->data());
+  }
+  return databento::Record{
+      reinterpret_cast<databento::RecordHeader*>(compat_buffer->data())};
+}
+}  // namespace
+
 databento::Record DbnDecoder::DecodeRecordCompat(
-    std::uint8_t version, VersionUpgradePolicy upgrade_policy,
+    std::uint8_t version, VersionUpgradePolicy upgrade_policy, bool ts_out,
     std::array<std::uint8_t, kMaxRecordLen>* compat_buffer, Record rec) {
   if (version == 1 && upgrade_policy == VersionUpgradePolicy::Upgrade) {
     if (rec.RType() == RType::InstrumentDef) {
-      auto v2 = rec.Get<InstrumentDefMsgV1>().ToV2();
-      auto v2_ptr = reinterpret_cast<std::uint8_t*>(&v2);
-      std::copy(v2_ptr, v2_ptr + v2.hd.Size(), compat_buffer->data());
-      return Record{reinterpret_cast<RecordHeader*>(compat_buffer->data())};
+      return UpgradeRecord<InstrumentDefMsgV1, InstrumentDefMsgV2>(
+          ts_out, compat_buffer, rec);
     } else if (rec.RType() == RType::SymbolMapping) {
-      auto v2 = rec.Get<SymbolMappingMsgV1>().ToV2();
-      auto v2_ptr = reinterpret_cast<std::uint8_t*>(&v2);
-      std::copy(v2_ptr, v2_ptr + v2.hd.Size(), compat_buffer->data());
-      return Record{reinterpret_cast<RecordHeader*>(compat_buffer->data())};
+      return UpgradeRecord<SymbolMappingMsgV1, SymbolMappingMsgV2>(
+          ts_out, compat_buffer, rec);
+    } else if (rec.RType() == RType::Error) {
+      return UpgradeRecord<ErrorMsgV1, ErrorMsgV2>(ts_out, compat_buffer, rec);
+    } else if (rec.RType() == RType::System) {
+      return UpgradeRecord<SystemMsgV1, SystemMsgV2>(ts_out, compat_buffer,
+                                                     rec);
     }
   }
   return rec;
@@ -228,7 +253,7 @@ const databento::Record* DbnDecoder::DecodeRecord() {
   current_record_ = Record{BufferRecordHeader()};
   buffer_idx_ += current_record_.Size();
   current_record_ = DbnDecoder::DecodeRecordCompat(
-      version_, upgrade_policy_, &compat_buffer_, current_record_);
+      version_, upgrade_policy_, ts_out_, &compat_buffer_, current_record_);
   return &current_record_;
 }
 

src/live_blocking.cpp

@@ -143,8 +143,9 @@ const databento::Record* LiveBlocking::NextRecord(
   }
   current_record_ = Record{BufferRecordHeader()};
   buffer_idx_ += current_record_.Size();
-  current_record_ = DbnDecoder::DecodeRecordCompat(
-      version_, upgrade_policy_, &compat_buffer_, current_record_);
+  current_record_ =
+      DbnDecoder::DecodeRecordCompat(version_, upgrade_policy_, send_ts_out_,
+                                     &compat_buffer_, current_record_);
   return &current_record_;
 }
 

test/src/dbn_decoder_tests.cpp

@@ -1,5 +1,6 @@
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <cstring>
@@ -9,6 +10,7 @@
 
 #include "databento/compat.hpp"
 #include "databento/constants.hpp"
+#include "databento/datetime.hpp"
 #include "databento/dbn.hpp"
 #include "databento/dbn_decoder.hpp"
 #include "databento/detail/file_stream.hpp"
@@ -18,6 +20,7 @@
 #include "databento/exceptions.hpp"
 #include "databento/ireadable.hpp"
 #include "databento/record.hpp"
+#include "databento/with_ts_out.hpp"
 
 namespace databento {
 namespace test {
@@ -138,6 +141,59 @@ TEST_F(DbnDecoderTests, TestDecodeDefinitionUpgrade) {
   AssertDefEq<InstrumentDefMsgV2>(ch_record2, f_record2);
 }
 
+TEST_F(DbnDecoderTests, TestUpgradeSymbolMappingWithTsOut) {
+  SymbolMappingMsgV1 sym_map{
+      {sizeof(SymbolMappingMsgV1) / RecordHeader::kLengthMultiplier,
+       RType::SymbolMapping, 0, 1, UnixNanos{std::chrono::nanoseconds{2}}},
+      {"ES.c.0"},
+      {"ESH4"},
+      {},
+      {},
+      {}};
+  WithTsOut<SymbolMappingMsgV1> orig{
+      sym_map, UnixNanos{std::chrono::system_clock::now()}};
+  std::array<std::uint8_t, kMaxRecordLen> compat_buffer{};
+  const auto res =
+      DbnDecoder::DecodeRecordCompat(1, VersionUpgradePolicy::Upgrade, true,
+                                     &compat_buffer, Record{&orig.rec.hd});
+  const auto& upgraded = res.Get<WithTsOut<SymbolMappingMsgV2>>();
+  ASSERT_EQ(orig.ts_out, upgraded.ts_out);
+  ASSERT_STREQ(orig.rec.STypeInSymbol(), upgraded.rec.STypeInSymbol());
+  ASSERT_STREQ(orig.rec.STypeOutSymbol(), upgraded.rec.STypeOutSymbol());
+  // `length` properly set
+  ASSERT_EQ(upgraded.rec.hd.Size(), sizeof(upgraded));
+  // used compat buffer
+  ASSERT_EQ(reinterpret_cast<const std::uint8_t*>(&upgraded),
+            compat_buffer.data());
+}
+
+TEST_F(DbnDecoderTests, TestUpgradeMbp1WithTsOut) {
+  WithTsOut<Mbp1Msg> orig{
+      Mbp1Msg{{sizeof(Mbp1Msg) / RecordHeader::kLengthMultiplier,
+               RType::Mbp1,
+               {},
+               {},
+               {}},
+              1'250'000'000,
+              {},
+              {},
+              Side::Ask,
+              {},
+              {},
+              {},
+              {},
+              {},
+              {}},
+      {std::chrono::system_clock::now()}};
+  std::array<std::uint8_t, kMaxRecordLen> compat_buffer{};
+  const auto res =
+      DbnDecoder::DecodeRecordCompat(1, VersionUpgradePolicy::Upgrade, true,
+                                     &compat_buffer, Record{&orig.rec.hd});
+  const auto& upgraded = res.Get<WithTsOut<Mbp1Msg>>();
+  // compat buffer unused and pointer unchanged
+  ASSERT_EQ(&orig, &upgraded);
+}
+
 class DbnDecoderSchemaTests
     : public DbnDecoderTests,
       public testing::WithParamInterface<std::pair<const char*, std::uint8_t>> {

test/src/live_blocking_tests.cpp

@@ -267,7 +267,7 @@ TEST_F(LiveBlockingTests, TestNextRecordWithTsOut) {
   constexpr auto kRecCount = 5;
   constexpr auto kTsOut = true;
   constexpr WithTsOut<TradeMsg> kRec{
-      {DummyHeader<WithTsOut<TradeMsg>>(RType::Mbp0),
+      {DummyHeader<TradeMsg>(RType::Mbp0),
        1,
        2,
        Action::Add,