FIX: Fix zstd get_range buffer management

Author: threecgreen

CHANGELOG.md

@@ -31,6 +31,11 @@ upgrading the input to DBNv3.
       has been updated to match
     - The previous `StatMsg` has been moved to `v2::StatMsg` or `StatMsgV2`
 
+### Bug fixes
+- Fixed "Zstd error decompressing: Operation made no progress over multiple calls, due
+  to output buffer being full" error with `TimeseriesGetRange`
+- Fixed missing implementation of `HistoricalBuilder::SetLogReceiver`
+
 ## 0.35.1 - 2025-05-20
 
 ### Bug fixes

examples/historical/timeseries_get_range.cpp

@@ -1,34 +1,15 @@
-#include <chrono>
-#include <ctime>
-#include <iomanip>
 #include <iostream>  // setw
-#ifdef _WIN32
-// _mkgmtime is equivalent to timegm
-#define timegm _mkgmtime
-#endif
 
 #include "databento/constants.hpp"
-#include "databento/datetime.hpp"
 #include "databento/enums.hpp"
 #include "databento/historical.hpp"
 
-// Converts a date to Unix Epoch nanoseconds
-databento::UnixNanos DateToUnixNanos(int year, int month, int day) {
-  std::tm time{};
-  time.tm_year = year - 1900;
-  // January is 0
-  time.tm_mon = month - 1;
-  time.tm_mday = day;
-  return databento::UnixNanos{std::chrono::seconds{::timegm(&time)}};
-}
-
 int main() {
   auto client = databento::HistoricalBuilder{}.SetKeyFromEnv().Build();
-  const databento::UnixNanos start = DateToUnixNanos(2022, 10, 3);
-  const databento::UnixNanos end = DateToUnixNanos(2022, 10, 4);
   const auto limit = 1000;
   client.TimeseriesGetRange(
-      databento::dataset::kGlbxMdp3, {start, end}, {"ESZ2"},
+      databento::dataset::kGlbxMdp3,
+      databento::DateTimeRange<std::string>{"2022-10-03"}, {"ESZ2"},
       databento::Schema::Trades, databento::SType::RawSymbol,
       databento::SType::InstrumentId, limit,
       [](databento::Metadata&& metadata) { std::cout << metadata << '\n'; },

include/databento/detail/buffer.hpp

@@ -3,6 +3,7 @@
 #include <cstddef>
 #include <memory>
 #include <new>
+#include <ostream>
 
 #include "databento/ireadable.hpp"
 #include "databento/iwritable.hpp"
@@ -22,22 +23,32 @@ class Buffer : public IReadable, public IWritable {
   void WriteAll(const char* data, std::size_t length);
   void WriteAll(const std::byte* data, std::size_t length) override;
 
-  std::byte*& WriteBegin() { return write_pos_; }
+  std::byte* WriteBegin() { return write_pos_; }
   std::byte* WriteEnd() { return end_; }
   const std::byte* WriteBegin() const { return write_pos_; }
   const std::byte* WriteEnd() const { return end_; }
+  // Indicate how many bytes were written
+  void Fill(std::size_t length) { write_pos_ += length; }
   std::size_t WriteCapacity() const {
     return static_cast<std::size_t>(end_ - write_pos_);
   }
 
-  /// Will throw if `length > ReadCapacity()`.
+  // Will throw if `length > ReadCapacity()`.
   void ReadExact(std::byte* buffer, std::size_t length) override;
   std::size_t ReadSome(std::byte* buffer, std::size_t max_length) override;
 
-  std::byte*& ReadBegin() { return read_pos_; }
+  std::byte* ReadBegin() { return read_pos_; }
   std::byte* ReadEnd() { return write_pos_; }
   const std::byte* ReadBegin() const { return read_pos_; }
   const std::byte* ReadEnd() const { return write_pos_; }
+  // Indicate how mnay bytes were read
+  void Consume(std::size_t length) {
+    read_pos_ += length;
+    if (static_cast<std::size_t>(read_pos_ - buf_.get()) > (Capacity() / 2)) {
+      Shift();
+    }
+  }
+  void ConsumeNoShift(std::size_t length) { read_pos_ += length; }
   std::size_t ReadCapacity() const {
     return static_cast<std::size_t>(write_pos_ - read_pos_);
   }
@@ -52,6 +63,8 @@ class Buffer : public IReadable, public IWritable {
   void Reserve(std::size_t capacity);
   void Shift();
 
+  friend std::ostream& operator<<(std::ostream& stream, const Buffer& buffer);
+
  private:
   static constexpr std::align_val_t kAlignment{8};
 

src/dbn_decoder.cpp

@@ -82,7 +82,7 @@ DbnDecoder::DbnDecoder(ILogReceiver* log_receiver,
     input_ =
         std::make_unique<detail::ZstdDecodeStream>(std::move(input_), buffer_);
     input_->ReadExact(buffer_.WriteBegin(), kMagicSize);
-    buffer_.WriteBegin() += kMagicSize;
+    buffer_.Fill(kMagicSize);
     const auto* buf_ptr = buffer_.ReadBegin();
     if (std::strncmp(Consume(buf_ptr, 3), kDbnPrefix, 3) != 0) {
       throw DbnResponseError{"Found Zstd input, but not DBN prefix"};
@@ -181,17 +181,17 @@ databento::Metadata DbnDecoder::DecodeMetadata() {
   // already read first 4 bytes detecting compression
   const auto read_size = kMetadataPreludeSize - kMagicSize;
   input_->ReadExact(buffer_.WriteBegin(), read_size);
-  buffer_.WriteBegin() += read_size;
+  buffer_.Fill(read_size);
   const auto [version, size] = DbnDecoder::DecodeMetadataVersionAndSize(
       buffer_.ReadBegin(), kMetadataPreludeSize);
-  buffer_.ReadBegin() += kMetadataPreludeSize;
+  buffer_.Consume(kMetadataPreludeSize);
   version_ = version;
   buffer_.Reserve(size);
   input_->ReadExact(buffer_.WriteBegin(), size);
-  buffer_.WriteBegin() += size;
+  buffer_.Fill(size);
   auto metadata = DbnDecoder::DecodeMetadataFields(
       version_, buffer_.ReadBegin(), buffer_.ReadEnd());
-  buffer_.ReadBegin() += size;
+  buffer_.Consume(size);
   // Metadata may leave buffer misaligned. Shift records to ensure 8-byte
   // alignment
   buffer_.Shift();
@@ -316,7 +316,7 @@ const databento::Record* DbnDecoder::DecodeRecord() {
     }
   }
   current_record_ = Record{BufferRecordHeader()};
-  buffer_.ReadBegin() += current_record_.Size();
+  buffer_.Consume(current_record_.Size());
   current_record_ = DbnDecoder::DecodeRecordCompat(
       version_, upgrade_policy_, ts_out_, &compat_buffer_, current_record_);
   return &current_record_;
@@ -328,7 +328,7 @@ size_t DbnDecoder::FillBuffer() {
   }
   const auto fill_size =
       input_->ReadSome(buffer_.WriteBegin(), buffer_.WriteCapacity());
-  buffer_.WriteBegin() += fill_size;
+  buffer_.Fill(fill_size);
   return fill_size;
 }
 
@@ -338,7 +338,7 @@ databento::RecordHeader* DbnDecoder::BufferRecordHeader() {
 
 bool DbnDecoder::DetectCompression() {
   input_->ReadExact(buffer_.WriteBegin(), kMagicSize);
-  buffer_.WriteBegin() += kMagicSize;
+  buffer_.Fill(kMagicSize);
   const auto* buffer_it = buffer_.ReadBegin();
   if (std::strncmp(Consume(buffer_it, 3), kDbnPrefix, 3) == 0) {
     return false;

src/detail/buffer.cpp

@@ -4,6 +4,7 @@
 #include <sstream>
 
 #include "databento/exceptions.hpp"
+#include "stream_op_helper.hpp"
 
 using databento::detail::Buffer;
 
@@ -16,7 +17,7 @@ size_t Buffer::Write(const std::byte* data, std::size_t length) {
   }
   const auto write_size = std::min(WriteCapacity(), length);
   std::copy(data, data + write_size, WriteBegin());
-  WriteBegin() += write_size;
+  Fill(length);
   return write_size;
 }
 
@@ -34,7 +35,7 @@ void Buffer::WriteAll(const std::byte* data, std::size_t length) {
 }
 
 void Buffer::ReadExact(std::byte* buffer, std::size_t length) {
-  if (length < ReadCapacity()) {
+  if (length > ReadCapacity()) {
     std::ostringstream err_msg;
     err_msg << "Reached end of buffer without " << length << " bytes, only "
             << ReadCapacity() << " bytes available";
@@ -46,7 +47,7 @@ void Buffer::ReadExact(std::byte* buffer, std::size_t length) {
 std::size_t Buffer::ReadSome(std::byte* buffer, std::size_t max_length) {
   const auto read_size = std::min(ReadCapacity(), max_length);
   std::copy(ReadBegin(), ReadBegin() + read_size, buffer);
-  ReadBegin() += read_size;
+  Consume(read_size);
   return read_size;
 }
 
@@ -71,3 +72,18 @@ void Buffer::Shift() {
   read_pos_ = buf_.get();
   write_pos_ = read_pos_ + unread_bytes;
 }
+
+std::ostream& databento::detail::operator<<(std::ostream& stream,
+                                            const Buffer& buffer) {
+  return StreamOpBuilder{stream}
+      .SetTypeName("Buffer")
+      .SetSpacer(" ")
+      .Build()
+      .AddField("buf_", buffer.buf_.get())
+      .AddField("end_", buffer.end_)
+      .AddField("read_pos", buffer.read_pos_)
+      .AddField("write_pos_", buffer.write_pos_)
+      .AddField("ReadCapacity", buffer.ReadCapacity())
+      .AddField("WriteCapacity", buffer.WriteCapacity())
+      .Finish();
+}

src/detail/dbn_buffer_decoder.cpp

@@ -13,7 +13,7 @@ databento::KeepGoing DbnBufferDecoder::Process(const char* data,
   zstd_buffer_->WriteAll(data, length);
   const auto read_size = zstd_stream_.ReadSome(dbn_buffer_.WriteBegin(),
                                                dbn_buffer_.WriteCapacity());
-  dbn_buffer_.WriteBegin() += read_size;
+  dbn_buffer_.Fill(read_size);
   if (read_size == 0) {
     return KeepGoing::Continue;
   }
@@ -25,7 +25,7 @@ databento::KeepGoing DbnBufferDecoder::Process(const char* data,
       std::tie(input_version_, bytes_needed_) =
           DbnDecoder::DecodeMetadataVersionAndSize(dbn_buffer_.ReadBegin(),
                                                    dbn_buffer_.ReadCapacity());
-      dbn_buffer_.ReadBegin() += kMetadataPreludeSize;
+      dbn_buffer_.Consume(kMetadataPreludeSize);
       dbn_buffer_.Reserve(bytes_needed_);
       state_ = DecoderState::Metadata;
       [[fallthrough]];
@@ -36,7 +36,7 @@ databento::KeepGoing DbnBufferDecoder::Process(const char* data,
       }
       auto metadata = DbnDecoder::DecodeMetadataFields(
           input_version_, dbn_buffer_.ReadBegin(), dbn_buffer_.ReadEnd());
-      dbn_buffer_.ReadBegin() += bytes_needed_;
+      dbn_buffer_.Consume(bytes_needed_);
       // Metadata may leave buffer misaligned. Shift records to ensure 8-byte
       // alignment
       dbn_buffer_.Shift();
@@ -61,7 +61,7 @@ databento::KeepGoing DbnBufferDecoder::Process(const char* data,
         if (record_callback_(record) == KeepGoing::Stop) {
           return KeepGoing::Stop;
         }
-        dbn_buffer_.ReadBegin() += bytes_needed_;
+        dbn_buffer_.Consume(bytes_needed_);
       }
     }
   }

src/detail/zstd_stream.cpp

@@ -24,7 +24,7 @@ ZstdDecodeStream::ZstdDecodeStream(std::unique_ptr<IReadable> input,
       read_suggestion_{::ZSTD_initDStream(z_dstream_.get())},
       in_buffer_{in_buffer.ReadBegin(), in_buffer.ReadEnd()},
       z_in_buffer_{in_buffer_.data(), in_buffer_.size(), 0} {
-  in_buffer.ReadBegin() += in_buffer.ReadCapacity();
+  in_buffer.Consume(in_buffer.ReadCapacity());
 }
 
 void ZstdDecodeStream::ReadExact(std::byte* buffer, std::size_t length) {

src/historical.cpp

@@ -949,6 +949,12 @@ HistoricalBuilder& HistoricalBuilder::SetGateway(HistoricalGateway gateway) {
   return *this;
 }
 
+HistoricalBuilder& HistoricalBuilder::SetLogReceiver(
+    ILogReceiver* log_receiver) {
+  log_receiver_ = log_receiver;
+  return *this;
+}
+
 Historical HistoricalBuilder::Build() {
   if (key_.empty()) {
     throw Exception{"'key' is unset"};

src/live_blocking.cpp

@@ -140,16 +140,16 @@ void LiveBlocking::Subscribe(std::string_view sub_msg,
 databento::Metadata LiveBlocking::Start() {
   client_.WriteAll("start_session\n");
   client_.ReadExact(buffer_.WriteBegin(), kMetadataPreludeSize);
-  buffer_.WriteBegin() += kMetadataPreludeSize;
+  buffer_.Fill(kMetadataPreludeSize);
   const auto [version, size] = DbnDecoder::DecodeMetadataVersionAndSize(
       buffer_.ReadBegin(), kMetadataPreludeSize);
-  buffer_.ReadBegin() += kMetadataPreludeSize;
+  buffer_.Consume(kMetadataPreludeSize);
   buffer_.Reserve(size);
   client_.ReadExact(buffer_.WriteBegin(), size);
-  buffer_.WriteBegin() += size;
+  buffer_.Fill(size);
   auto metadata = DbnDecoder::DecodeMetadataFields(version, buffer_.ReadBegin(),
                                                    buffer_.ReadEnd());
-  buffer_.ReadBegin() += size;
+  buffer_.Consume(size);
   // Metadata may leave buffer misaligned. Shift records to ensure 8-byte
   // alignment
   buffer_.Shift();
@@ -184,7 +184,7 @@ const databento::Record* LiveBlocking::NextRecord(
     }
   }
   current_record_ = Record{BufferRecordHeader()};
-  buffer_.ReadBegin() += current_record_.Size();
+  buffer_.Consume(current_record_.Size());
   current_record_ =
       DbnDecoder::DecodeRecordCompat(version_, upgrade_policy_, send_ts_out_,
                                      &compat_buffer_, current_record_);
@@ -223,7 +223,7 @@ std::string LiveBlocking::DecodeChallenge() {
   if (read_size == 0) {
     throw LiveApiError{"Gateway closed socket during authentication"};
   }
-  buffer_.WriteBegin() += read_size;
+  buffer_.Fill(read_size);
   // first line is version
   std::string response{reinterpret_cast<const char*>(buffer_.ReadBegin()),
                        buffer_.ReadCapacity()};
@@ -243,9 +243,8 @@ std::string LiveBlocking::DecodeChallenge() {
                          : response.find('\n', find_start);
   while (next_nl_pos == std::string::npos) {
     // read more
-    buffer_.WriteBegin() +=
-        client_.ReadSome(buffer_.WriteBegin(), buffer_.WriteCapacity())
-            .read_size;
+    buffer_.Fill(client_.ReadSome(buffer_.WriteBegin(), buffer_.WriteCapacity())
+                     .read_size);
     if (buffer_.ReadCapacity() == 0) {
       throw LiveApiError{"Gateway closed socket during authentication"};
     }
@@ -335,7 +334,7 @@ std::uint64_t LiveBlocking::DecodeAuthResp() {
           "Unexpected end of message received from server after replying to "
           "CRAM"};
     }
-    buffer_.WriteBegin() += read_size;
+    buffer_.Fill(read_size);
     newline_ptr = std::find(buffer_.ReadBegin(), buffer_.ReadEnd(),
                             static_cast<std::byte>('\n'));
   } while (newline_ptr == buffer_.ReadEnd());
@@ -349,7 +348,7 @@ std::uint64_t LiveBlocking::DecodeAuthResp() {
     log_receiver_->Receive(LogLevel::Debug, log_ss.str());
   }
   // set in case Read call also read records. One beyond newline
-  buffer_.ReadBegin() += response.length() + 1;
+  buffer_.Consume(response.length() + 1);
 
   std::size_t pos{};
   bool found_success{};
@@ -411,7 +410,7 @@ databento::detail::TcpClient::Result LiveBlocking::FillBuffer(
   buffer_.Shift();
   const auto read_res =
       client_.ReadSome(buffer_.WriteBegin(), buffer_.WriteCapacity(), timeout);
-  buffer_.WriteBegin() += read_res.read_size;
+  buffer_.Fill(read_res.read_size);
   return read_res;
 }
 

tests/src/buffer_tests.cpp

@@ -10,8 +10,8 @@ using namespace std::string_view_literals;
 namespace databento::detail::tests {
 TEST(BufferTests, TestWriteAllPastCapacity) {
   Buffer target{10};
-  target.WriteBegin() += 4;
-  target.ReadBegin() += 2;
+  target.Fill(4);
+  target.ConsumeNoShift(2);
   ASSERT_EQ(target.WriteCapacity(), 6);
   ASSERT_EQ(target.ReadCapacity(), 2);
   ASSERT_EQ(target.Capacity(), 10);
@@ -25,7 +25,7 @@ TEST(BufferTests, TestWriteAllPastCapacity) {
 TEST(BufferTests, TestWriteAllShift) {
   Buffer target{20};
   target.WriteAll("TestWriteAllShift", 17);
-  target.ReadBegin() += 4;
+  target.ConsumeNoShift(4);
   ASSERT_EQ(target.WriteCapacity(), 3);
   ASSERT_EQ(target.ReadCapacity(), 13);
   ASSERT_EQ(target.Capacity(), 20);
@@ -38,8 +38,8 @@ TEST(BufferTests, TestWriteAllShift) {
 
 TEST(BufferTests, TestWriteRead) {
   Buffer target{10};
-  target.WriteBegin() += 5;
-  target.ReadBegin() += 5;
+  target.Fill(5);
+  target.ConsumeNoShift(5);
   const auto write_len = target.Write("BufferTests", 11);
   ASSERT_EQ(write_len, 10);
   std::array<std::byte, 10> read_buf{};
@@ -55,7 +55,16 @@ TEST(BufferTests, TestReserve) {
   ASSERT_EQ(target.ReadCapacity(), 0);
   ASSERT_EQ(target.Capacity(), 120);
   target.WriteAll("TestReserve", 11);
-  target.ReadBegin() += 4;
+  target.ConsumeNoShift(4);
 }
 
+TEST(BufferTests, TestConsumeShift) {
+  Buffer target{120};
+  target.Fill(120);
+  ASSERT_EQ(target.WriteCapacity(), 0);
+  target.ConsumeNoShift(100);
+  ASSERT_EQ(target.WriteCapacity(), 0);
+  target.Consume(1);
+  ASSERT_EQ(target.WriteCapacity(), 101);
+}
 }  // namespace databento::detail::tests