FIX: Fix Zstd decoding and batch download in C++

Author: threecgreen

.github/workflows/build.yaml

@@ -22,7 +22,7 @@ jobs:
       - name: Install dependencies
         run: |
           sudo apt-get update
-          sudo apt-get install libzstd-dev ninja-build
+          sudo apt-get install cppcheck libzstd-dev ninja-build
       - name: Install gtest
         uses: MarkusJx/googletest-installer@v1.1
       - name: CMake configure
@@ -33,6 +33,7 @@ jobs:
             -DDATABENTO_ENABLE_UNIT_TESTING=1 \
             -DDATABENTO_ENABLE_EXAMPLES=1 \
             -DDATABENTO_ENABLE_CLANG_TIDY=1 \
+            -DDATABENTO_ENABLE_CPPCHECK=1 \
             -DDATABENTO_ENABLE_ASAN=1 \
             -DDATABENTO_ENABLE_UBSAN=1
       - name: CMake build
@@ -48,16 +49,16 @@ jobs:
       - name: Checkout repository
         uses: actions/checkout@v3
       - name: Install dependencies
-        run: brew install cmake googletest openssl@3 ninja zstd
+        run: brew install cmake cppcheck googletest openssl@3 ninja zstd
       # Don't enable clang-tidy on macOS because it's incredibly slow
       - name: CMake configure
         run: |
           cmake -S . -B build \
             -GNinja \
             -DDATABENTO_ENABLE_UNIT_TESTING=1 \
             -DDATABENTO_ENABLE_EXAMPLES=1 \
-            -DDATABENTO_ENABLE_ASAN=1 \
-            -DDATABENTO_ENABLE_UBSAN=1 \
+            -DDATABENTO_ENABLE_CPPCHECK=1 \
+            -DDATABENTO_ENABLE_TSAN=1 \
             -DOPENSSL_ROOT_DIR=$(brew --prefix openssl@3)
       - name: CMake build
         run: cmake --build build

CHANGELOG.md

@@ -4,9 +4,13 @@
 
 #### Enhancements
 - Added constants for dataset codes for Databento Equity Basic and OPRA Pillar
+- Added `const char*` getters to records for fixed-length `char` arrays
+- Added `RType` getter to `Record`
 
 #### Bug fixes
 - Batch live subscriptions to avoid hitting max message length
+- Fix bug in Zstd decompression
+- Fix `Historical::BatchDownload` truncating file before writing each chunk
 
 ## 0.9.0 - 2023-06-13
 

CMakeLists.txt

@@ -135,6 +135,8 @@ else()
     )
   endif()
   FetchContent_MakeAvailable(json)
+  # Ignore compiler warnings in headers
+  add_system_include_property(nlohmann_json)
 endif()
 # cpp-httplib
 set(httplib_version 0.11.4)

cmake/StaticAnalyzers.cmake

@@ -17,19 +17,13 @@ if(${PROJECT_NAME_UPPERCASE}_ENABLE_CPPCHECK)
     set(
       CMAKE_CXX_CPPCHECK ${CPPCHECK}
       --enable=all
-      --suppress=missingIncludeSystem
+      --suppress=missingIncludeSystem  # False positives
       --suppress=preprocessorErrorDirective
-      --suppress=unusedFunction
-      --suppress=unmatchedSuppression
+      --suppress=unusedFunction        # False positives
+      --suppress=unmatchedSuppression  # Support different cppcheck versions
       --inline-suppr
       --relative-paths=${CMAKE_SOURCE_DIR}
       --quiet
-      -I${CMAKE_SOURCE_DIR}/include
-      -I${CMAKE_SOURCE_DIR}/src
-      # Ignore tests
-      -i${CMAKE_SOURCE_DIR}/test
-      # Ignore third-party dependencies
-      -i${CMAKE_BINARY_DIR}
     )
     message(STATUS "Cppcheck finished setting up.")
   else()

include/databento/historical.hpp

@@ -237,6 +237,8 @@ class Historical {
   using HttplibParams = std::multimap<std::string, std::string>;
 
   BatchJob BatchSubmitJob(const HttplibParams& params);
+  void StreamToFile(const std::string& url_path, const HttplibParams& params,
+                    const std::string& file_path);
   void DownloadFile(const std::string& url, const std::string& output_path);
   std::vector<BatchJob> BatchListJobs(const HttplibParams& params);
   std::vector<DatasetConditionDetail> MetadataGetDatasetCondition(

include/databento/record.hpp

@@ -36,6 +36,7 @@ class Record {
   explicit Record(RecordHeader* record) : record_{record} {}
 
   const RecordHeader& Header() const { return *record_; }
+  ::databento::RType RType() const { return record_->rtype; }
 
   template <typename T>
   bool Holds() const {
@@ -53,7 +54,7 @@ class Record {
 
   std::size_t Size() const;
   static std::size_t SizeOfSchema(Schema schema);
-  static RType RTypeFromSchema(Schema schema);
+  static ::databento::RType RTypeFromSchema(Schema schema);
 
  private:
   RecordHeader* record_;
@@ -168,6 +169,21 @@ static_assert(sizeof(OhlcvMsg) == 56, "OhlcvMsg size must match C");
 struct InstrumentDefMsg {
   static bool HasRType(RType rtype) { return rtype == RType::InstrumentDef; }
 
+  const char* Currency() const { return currency.data(); }
+  const char* SettlCurrency() const { return settl_currency.data(); }
+  const char* SecSubType() const { return secsubtype.data(); }
+  const char* RawSymbol() const { return raw_symbol.data(); }
+  const char* Group() const { return group.data(); }
+  const char* Exchange() const { return exchange.data(); }
+  const char* Asset() const { return asset.data(); }
+  const char* Cfi() const { return cfi.data(); }
+  const char* SecurityType() const { return security_type.data(); }
+  const char* UnitOfMeasure() const { return unit_of_measure.data(); }
+  const char* Underlying() const { return underlying.data(); }
+  const char* StrikePriceCurrency() const {
+    return strike_price_currency.data();
+  }
+
   RecordHeader hd;
   UnixNanos ts_recv;
   std::int64_t min_price_increment;
@@ -308,6 +324,9 @@ struct ErrorMsg {
 struct SymbolMappingMsg {
   static bool HasRType(RType rtype) { return rtype == RType::SymbolMapping; }
 
+  const char* STypeInSymbol() const { return stype_in_symbol.data(); }
+  const char* STypeOutSymbol() const { return stype_out_symbol.data(); }
+
   RecordHeader hd;
   std::array<char, 22> stype_in_symbol;
   std::array<char, 22> stype_out_symbol;

src/dbn_decoder.cpp

@@ -151,7 +151,7 @@ databento::Metadata DbnDecoder::DecodeMetadata() {
   const auto version_and_size =
       DbnDecoder::DecodeMetadataVersionAndSize(buffer_.data(), 8);
   buffer_.resize(version_and_size.second);
-  input_->ReadExact(buffer_.data(), version_and_size.second);
+  input_->ReadExact(buffer_.data(), buffer_.size());
   buffer_idx_ = buffer_.size();
   return DbnDecoder::DecodeMetadataFields(version_and_size.first, buffer_);
 }

src/detail/json_helpers.cpp

@@ -96,12 +96,7 @@ std::vector<std::string> ParseAt(const std::string& endpoint,
   if (!symbols_json.is_array()) {
     throw JsonResponseError::TypeMismatch(endpoint, key + " array", json);
   }
-  std::vector<std::string> res;
-  res.reserve(symbols_json.size());
-  for (const auto& item : symbols_json.items()) {
-    res.emplace_back(item.value());
-  }
-  return res;
+  return {symbols_json.begin(), symbols_json.end()};
 }
 
 }  // namespace detail

src/detail/zstd_stream.cpp

@@ -21,7 +21,7 @@ ZstdStream::ZstdStream(std::unique_ptr<IReadable> input,
 void ZstdStream::ReadExact(std::uint8_t* buffer, std::size_t length) {
   std::size_t size{};
   do {
-    size += ReadSome(buffer, length - size);
+    size += ReadSome(&buffer[size], length - size);
   } while (size < length && read_suggestion_ != 0);
   // check for end of stream without obtaining `length` bytes
   if (size < length) {

src/historical.cpp

@@ -10,6 +10,7 @@
 #include <exception>  // exception, exception_ptr
 #include <fstream>    // ofstream
 #include <ios>        // ios::binary
+#include <iterator>   // back_inserter
 #include <memory>     // unique_ptr
 #include <string>
 #include <utility>  // move
@@ -245,10 +246,8 @@ std::vector<databento::BatchJob> Historical::BatchListJobs(
     throw JsonResponseError::TypeMismatch(kEndpoint, "array", json);
   }
   std::vector<BatchJob> jobs;
-  jobs.reserve(json.size());
-  for (const auto& job_json : json.items()) {
-    jobs.emplace_back(::Parse(kEndpoint, job_json.value()));
-  }
+  std::transform(json.begin(), json.end(), std::back_inserter(jobs),
+                 [](const auto& item) { return ::Parse(kEndpoint, item); });
   return jobs;
 }
 
@@ -316,6 +315,21 @@ std::string Historical::BatchDownload(const std::string& output_dir,
   return output_path;
 }
 
+void Historical::StreamToFile(const std::string& url_path,
+                              const HttplibParams& params,
+                              const std::string& file_path) {
+  std::ofstream out_file{file_path, std::ios::binary};
+  if (out_file.fail()) {
+    throw InvalidArgumentError{"Historical::StreamToFile", "file_path",
+                               "Failed to open file"};
+  }
+  this->client_.GetRawStream(
+      url_path, params, [&out_file](const char* data, std::size_t length) {
+        out_file.write(data, static_cast<std::streamsize>(length));
+        return true;
+      });
+}
+
 void Historical::DownloadFile(const std::string& url,
                               const std::string& output_path) {
   static const std::string kEndpoint = "Historical::BatchDownload";
@@ -338,12 +352,7 @@ void Historical::DownloadFile(const std::string& url,
     path = url.substr(slash);
   }
 
-  client_.GetRawStream(
-      path, {}, [&output_path](const char* data, std::size_t length) {
-        std::ofstream out_file{output_path};
-        out_file.write(data, static_cast<std::streamsize>(length));
-        return KeepGoing::Continue;
-      });
+  StreamToFile(path, {}, output_path);
 }
 
 std::map<std::string, std::int32_t> Historical::MetadataListPublishers() {
@@ -848,14 +857,15 @@ databento::SymbologyResolution Historical::SymbologyResolve(
                                             mapping_json);
     }
     std::vector<StrMappingInterval> mapping_intervals;
-    mapping_intervals.reserve(mapping_json.size());
-    for (const auto& interval_json : mapping_json.items()) {
-      mapping_intervals.emplace_back(StrMappingInterval{
-          detail::CheckedAt(kEndpoint, interval_json.value(), "d0"),
-          detail::CheckedAt(kEndpoint, interval_json.value(), "d1"),
-          detail::CheckedAt(kEndpoint, interval_json.value(), "s"),
-      });
-    }
+    std::transform(mapping_json.begin(), mapping_json.end(),
+                   std::back_inserter(mapping_intervals),
+                   [](const auto& interval_json) {
+                     return StrMappingInterval{
+                         detail::CheckedAt(kEndpoint, interval_json, "d0"),
+                         detail::CheckedAt(kEndpoint, interval_json, "d1"),
+                         detail::CheckedAt(kEndpoint, interval_json, "s"),
+                     };
+                   });
     res.mappings.emplace(mapping.key(), std::move(mapping_intervals));
   }
   if (!partial_json.is_array()) {
@@ -1053,19 +1063,7 @@ databento::DbnFileStore Historical::TimeseriesGetRangeToFile(
 }
 databento::DbnFileStore Historical::TimeseriesGetRangeToFile(
     const HttplibParams& params, const std::string& file_path) {
-  {
-    std::ofstream out_file{file_path, std::ios::binary};
-    if (out_file.fail()) {
-      throw InvalidArgumentError{kTimeseriesGetRangeEndpoint, "file_path",
-                                 "Non-existent or invalid file"};
-    }
-    this->client_.GetRawStream(
-        kTimeseriesGetRangePath, params,
-        [&out_file](const char* data, std::size_t length) {
-          out_file.write(data, static_cast<std::streamsize>(length));
-          return true;
-        });
-  }  // close out_file
+  StreamToFile(kTimeseriesGetRangePath, params, file_path);
   return DbnFileStore{file_path};
 }