ADD: Increase SystemMsg and ErrorMsg size

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.15.0 - TBD
+
+### Breaking changes
+- Increased size of `SystemMsg` and `ErrorMsg` to provide better messages from Live
+  gateway
+  - Increased length of `err` and `msg` fields for more detailed messages
+  - Added `is_last` field to `ErrorMsg` to indicate the last error in a chain
+  - Added `code` field to `SystemMsg` and `ErrorMsg`, although currently unused
+  - Added new `is_last` parameter to `ErrorMsg::new`
+  - Decoding these is backwards-compatible and records with longer messages won't be
+    sent during the DBN version 2 migration period
+  - Renamed previous records to `ErrorMsgV1` and `SystemMsgV1`
+
 ## 0.14.1 - 2023-12-18
 
 ### Enhancements

include/databento/compat.hpp

@@ -18,7 +18,9 @@ constexpr std::size_t VersionSymbolCstrLen(std::uint8_t version) {
 }
 
 using InstrumentDefMsgV2 = InstrumentDefMsg;
+using ErrorMsgV2 = ErrorMsg;
 using SymbolMappingMsgV2 = SymbolMappingMsg;
+using SystemMsgV2 = SystemMsg;
 
 // DBN version 1 instrument definition.
 struct InstrumentDefMsgV1 {
@@ -111,6 +113,21 @@ struct InstrumentDefMsgV1 {
 static_assert(sizeof(InstrumentDefMsgV1) == 360, "Size must match Rust");
 static_assert(alignof(InstrumentDefMsgV1) == 8, "Must have 8-byte alignment");
 
+// An error message from the Live Subscription Gateway (LSG). This will never
+// be present in historical data.
+struct ErrorMsgV1 {
+  static bool HasRType(RType rtype) { return rtype == RType::Error; }
+
+  ErrorMsgV2 ToV2() const;
+  UnixNanos IndexTs() const { return hd.ts_event; }
+  const char* Err() const { return err.data(); }
+
+  RecordHeader hd;
+  std::array<char, 64> err;
+};
+static_assert(sizeof(ErrorMsgV1) == 80, "ErrorMsg size must match Rust");
+static_assert(alignof(ErrorMsgV1) == 8, "Must have 8-byte alignment");
+
 /// A symbol mapping message.
 struct SymbolMappingMsgV1 {
   static bool HasRType(RType rtype) { return rtype == RType::SymbolMapping; }
@@ -130,11 +147,33 @@ struct SymbolMappingMsgV1 {
 static_assert(sizeof(SymbolMappingMsgV1) == 80, "Size must match Rust");
 static_assert(alignof(SymbolMappingMsgV1) == 8, "Must have 8-byte alignment");
 
+struct SystemMsgV1 {
+  static bool HasRType(RType rtype) { return rtype == RType::System; }
+
+  SystemMsgV2 ToV2() const;
+  UnixNanos IndexTs() const { return hd.ts_event; }
+  const char* Msg() const { return msg.data(); }
+  bool IsHeartbeat() const {
+    return std::strncmp(msg.data(), "Heartbeat", 9) == 0;
+  }
+
+  RecordHeader hd;
+  std::array<char, 64> msg;
+};
+static_assert(sizeof(SystemMsgV1) == 80, "SystemMsg size must match Rust");
+static_assert(alignof(SystemMsgV1) == 8, "Must have 8-byte alignment");
+
 bool operator==(const InstrumentDefMsgV1& lhs, const InstrumentDefMsgV1& rhs);
 inline bool operator!=(const InstrumentDefMsgV1& lhs,
                        const InstrumentDefMsgV1& rhs) {
   return !(lhs == rhs);
 }
+inline bool operator==(const ErrorMsgV1& lhs, const ErrorMsgV1& rhs) {
+  return std::tie(lhs.hd, lhs.err) == std::tie(rhs.hd, rhs.err);
+}
+inline bool operator!=(const ErrorMsgV1& lhs, const ErrorMsgV1& rhs) {
+  return !(lhs == rhs);
+}
 inline bool operator==(const SymbolMappingMsgV1& lhs,
                        const SymbolMappingMsgV1& rhs) {
   return std::tie(lhs.hd, lhs.stype_in_symbol, lhs.stype_out_symbol,
@@ -146,10 +185,20 @@ inline bool operator!=(const SymbolMappingMsgV1& lhs,
                        const SymbolMappingMsgV1& rhs) {
   return !(lhs == rhs);
 }
+inline bool operator==(const SystemMsgV1& lhs, const SystemMsgV1& rhs) {
+  return std::tie(lhs.hd, lhs.msg) == std::tie(rhs.hd, rhs.msg);
+}
+inline bool operator!=(const SystemMsgV1& lhs, const SystemMsgV1& rhs) {
+  return !(lhs == rhs);
+}
 std::string ToString(const InstrumentDefMsgV1& instr_def_msg);
 std::ostream& operator<<(std::ostream& stream,
                          const InstrumentDefMsgV1& instr_def_msg);
+std::string ToString(const ErrorMsgV1& err_msg);
+std::ostream& operator<<(std::ostream& stream, const ErrorMsgV1& err_msg);
 std::string ToString(const SymbolMappingMsgV1& symbol_mapping_msg);
 std::ostream& operator<<(std::ostream& stream,
                          const SymbolMappingMsgV1& symbol_mapping_msg);
+std::string ToString(const SystemMsgV1& sys_msg);
+std::ostream& operator<<(std::ostream& stream, const SystemMsgV1& sys_msg);
 }  // namespace databento

include/databento/record.hpp

@@ -337,9 +337,11 @@ struct ErrorMsg {
   const char* Err() const { return err.data(); }
 
   RecordHeader hd;
-  std::array<char, 64> err;
+  std::array<char, 302> err;
+  std::uint8_t code;
+  std::uint8_t is_last;
 };
-static_assert(sizeof(ErrorMsg) == 80, "ErrorMsg size must match Rust");
+static_assert(sizeof(ErrorMsg) == 320, "ErrorMsg size must match Rust");
 static_assert(alignof(ErrorMsg) == 8, "Must have 8-byte alignment");
 
 /// A symbol mapping message.
@@ -372,9 +374,10 @@ struct SystemMsg {
   }
 
   RecordHeader hd;
-  std::array<char, 64> msg;
+  std::array<char, 303> msg;
+  std::uint8_t code;
 };
-static_assert(sizeof(SystemMsg) == 80, "SystemMsg size must match Rust");
+static_assert(sizeof(SystemMsg) == 320, "SystemMsg size must match Rust");
 static_assert(alignof(SystemMsg) == 8, "Must have 8-byte alignment");
 
 inline bool operator==(const RecordHeader& lhs, const RecordHeader& rhs) {
@@ -480,14 +483,15 @@ inline bool operator!=(const StatMsg& lhs, const StatMsg& rhs) {
 }
 
 inline bool operator==(const ErrorMsg& lhs, const ErrorMsg& rhs) {
-  return lhs.hd == rhs.hd && lhs.err == rhs.err;
+  return lhs.hd == rhs.hd && lhs.err == rhs.err && lhs.code == rhs.code &&
+         lhs.is_last == rhs.is_last;
 }
 inline bool operator!=(const ErrorMsg& lhs, const ErrorMsg& rhs) {
   return !(lhs == rhs);
 }
 
 inline bool operator==(const SystemMsg& lhs, const SystemMsg& rhs) {
-  return lhs.hd == rhs.hd && lhs.msg == rhs.msg;
+  return lhs.hd == rhs.hd && lhs.msg == rhs.msg && lhs.code == rhs.code;
 }
 inline bool operator!=(const SystemMsg& lhs, const SystemMsg& rhs) {
   return !(lhs == rhs);

src/compat.cpp

@@ -92,6 +92,17 @@ InstrumentDefMsgV2 InstrumentDefMsgV1::ToV2() const {
   return ret;
 }
 
+ErrorMsgV2 ErrorMsgV1::ToV2() const {
+  ErrorMsgV2 ret{
+      RecordHeader{sizeof(ErrorMsgV2) / RecordHeader::kLengthMultiplier,
+                   RType::Error, hd.publisher_id, hd.instrument_id},
+      {},
+      std::numeric_limits<std::uint8_t>::max(),
+      std::numeric_limits<std::uint8_t>::max()};
+  std::copy(err.begin(), err.end(), ret.err.begin());
+  return ret;
+}
+
 SymbolMappingMsgV2 SymbolMappingMsgV1::ToV2() const {
   SymbolMappingMsgV2 ret{
       RecordHeader{sizeof(SymbolMappingMsgV2) / RecordHeader::kLengthMultiplier,
@@ -111,6 +122,16 @@ SymbolMappingMsgV2 SymbolMappingMsgV1::ToV2() const {
   return ret;
 }
 
+SystemMsgV2 SystemMsgV1::ToV2() const {
+  SystemMsgV2 ret{
+      RecordHeader{sizeof(SystemMsgV2) / RecordHeader::kLengthMultiplier,
+                   RType::System, hd.publisher_id, hd.instrument_id},
+      {},
+      std::numeric_limits<std::uint8_t>::max()};
+  std::copy(msg.begin(), msg.end(), ret.msg.begin());
+  return ret;
+}
+
 bool operator==(const InstrumentDefMsgV1& lhs, const InstrumentDefMsgV1& rhs) {
   return lhs.hd == rhs.hd && lhs.ts_recv == rhs.ts_recv &&
          lhs.min_price_increment == rhs.min_price_increment &&
@@ -245,6 +266,16 @@ std::ostream& operator<<(std::ostream& stream,
       .AddField("tick_rule", instr_def_msg.tick_rule)
       .Finish();
 }
+std::string ToString(const ErrorMsgV1& err_msg) { return MakeString(err_msg); }
+std::ostream& operator<<(std::ostream& stream, const ErrorMsgV1& err_msg) {
+  return StreamOpBuilder{stream}
+      .SetSpacer("\n    ")
+      .SetTypeName("ErrorMsgV1")
+      .Build()
+      .AddField("hd", err_msg.hd)
+      .AddField("err", err_msg.err)
+      .Finish();
+}
 std::string ToString(const SymbolMappingMsgV1& symbol_mapping_msg) {
   return MakeString(symbol_mapping_msg);
 }
@@ -261,4 +292,16 @@ std::ostream& operator<<(std::ostream& stream,
       .AddField("end_ts", symbol_mapping_msg.end_ts)
       .Finish();
 }
+std::string ToString(const SystemMsgV1& system_msg) {
+  return MakeString(system_msg);
+}
+std::ostream& operator<<(std::ostream& stream, const SystemMsgV1& system_msg) {
+  return StreamOpBuilder{stream}
+      .SetSpacer("\n    ")
+      .SetTypeName("SystemMsgV1")
+      .Build()
+      .AddField("hd", system_msg.hd)
+      .AddField("msg", system_msg.msg)
+      .Finish();
+}
 }  // namespace databento

src/record.cpp

@@ -467,6 +467,8 @@ std::ostream& operator<<(std::ostream& stream, const ErrorMsg& err_msg) {
       .Build()
       .AddField("hd", err_msg.hd)
       .AddField("err", err_msg.err)
+      .AddField("code", err_msg.code)
+      .AddField("is_last", err_msg.is_last)
       .Finish();
 }
 
@@ -480,6 +482,7 @@ std::ostream& operator<<(std::ostream& stream, const SystemMsg& system_msg) {
       .Build()
       .AddField("hd", system_msg.hd)
       .AddField("msg", system_msg.msg)
+      .AddField("code", system_msg.code)
       .Finish();
 }