ADD: Add slow reader behavior support to clients

threecgreen · threecgreen · commit 478f2a20c78e · 2026-02-17T14:46:49.000Z
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,11 @@
 
 ## 0.47.1 - TBD
 
+### Enhancements
+- Added `SlowReadBehavior` enum and `LiveBuilder::SetSlowReadBehavior()` to configure
+  gateway behavior when client falls behind
+- Added `SlowReadBehavior()` getter to `LiveBlocking` and `LiveThreaded`
+
 ### Bug fixes
 - Added conversion for missing schemas for function `RTypeFromSchema`
 
diff --git a/include/databento/enums.hpp b/include/databento/enums.hpp
@@ -48,6 +48,15 @@ enum class DatasetCondition : std::uint8_t {
   Missing,
 };
 
+// Live session parameter which controls gateway behavior when the client
+// falls behind real time.
+enum class SlowReadBehavior : std::uint8_t {
+  // Send a warning but continue reading.
+  Warn = 0,
+  // Skip records to catch up.
+  Skip = 1,
+};
+
 // A record type sentinel.
 namespace r_type {
 enum RType : std::uint8_t {
@@ -662,6 +671,7 @@ const char* ToString(SplitDuration duration_interval);
 const char* ToString(Delivery delivery);
 const char* ToString(JobState state);
 const char* ToString(DatasetCondition condition);
+const char* ToString(SlowReadBehavior slow_read_behavior);
 const char* ToString(RType r_type);
 const char* ToString(Side side);
 const char* ToString(Action action);
@@ -688,6 +698,7 @@ std::ostream& operator<<(std::ostream& out, SplitDuration duration_interval);
 std::ostream& operator<<(std::ostream& out, Delivery delivery);
 std::ostream& operator<<(std::ostream& out, JobState state);
 std::ostream& operator<<(std::ostream& out, DatasetCondition condition);
+std::ostream& operator<<(std::ostream& out, SlowReadBehavior slow_read_behavior);
 std::ostream& operator<<(std::ostream& out, RType r_type);
 std::ostream& operator<<(std::ostream& out, Side side);
 std::ostream& operator<<(std::ostream& out, Action action);
diff --git a/include/databento/live.hpp b/include/databento/live.hpp
@@ -2,6 +2,8 @@
 
 #include <chrono>
 #include <cstddef>
+#include <cstdint>
+#include <optional>
 #include <string>
 
 #include "databento/enums.hpp"  // VersionUpgradePolicy
@@ -53,6 +55,8 @@ class LiveBuilder {
   LiveBuilder& ExtendUserAgent(std::string extension);
   // Sets the compression mode for the read stream.
   LiveBuilder& SetCompression(Compression compression);
+  // Sets the behavior of the gateway when the client falls behind real time.
+  LiveBuilder& SetSlowReadBehavior(SlowReadBehavior slow_read_behavior);
 
   /*
    * Build a live client instance
@@ -80,5 +84,6 @@ class LiveBuilder {
   std::size_t buffer_size_;
   std::string user_agent_ext_;
   Compression compression_{Compression::None};
+  std::optional<SlowReadBehavior> slow_read_behavior_{};
 };
 }  // namespace databento
diff --git a/include/databento/live_blocking.hpp b/include/databento/live_blocking.hpp
@@ -45,6 +45,9 @@ class LiveBlocking {
     return heartbeat_interval_;
   }
   databento::Compression Compression() const { return compression_; }
+  std::optional<databento::SlowReadBehavior> SlowReadBehavior() const {
+    return slow_read_behavior_;
+  }
   const std::vector<LiveSubscription>& Subscriptions() const { return subscriptions_; }
   std::vector<LiveSubscription>& Subscriptions() { return subscriptions_; }
 
@@ -95,13 +98,15 @@ class LiveBlocking {
                bool send_ts_out, VersionUpgradePolicy upgrade_policy,
                std::optional<std::chrono::seconds> heartbeat_interval,
                std::size_t buffer_size, std::string user_agent_ext,
-               databento::Compression compression);
+               databento::Compression compression,
+               std::optional<databento::SlowReadBehavior> slow_read_behavior);
   LiveBlocking(ILogReceiver* log_receiver, std::string key, std::string dataset,
                std::string gateway, std::uint16_t port, bool send_ts_out,
                VersionUpgradePolicy upgrade_policy,
                std::optional<std::chrono::seconds> heartbeat_interval,
                std::size_t buffer_size, std::string user_agent_ext,
-               databento::Compression compression);
+               databento::Compression compression,
+               std::optional<databento::SlowReadBehavior> slow_read_behavior);
 
   std::string DetermineGateway() const;
   std::uint64_t Authenticate();
@@ -128,6 +133,7 @@ class LiveBlocking {
   const VersionUpgradePolicy upgrade_policy_;
   const std::optional<std::chrono::seconds> heartbeat_interval_;
   const databento::Compression compression_;
+  const std::optional<databento::SlowReadBehavior> slow_read_behavior_;
   detail::LiveConnection connection_;
   std::uint32_t sub_counter_{};
   std::vector<LiveSubscription> subscriptions_;
diff --git a/include/databento/live_threaded.hpp b/include/databento/live_threaded.hpp
@@ -55,6 +55,7 @@ class LiveThreaded {
   VersionUpgradePolicy UpgradePolicy() const;
   std::optional<std::chrono::seconds> HeartbeatInterval() const;
   databento::Compression Compression() const;
+  std::optional<databento::SlowReadBehavior> SlowReadBehavior() const;
   const std::vector<LiveSubscription>& Subscriptions() const;
   std::vector<LiveSubscription>& Subscriptions();
 
@@ -109,13 +110,15 @@ class LiveThreaded {
                bool send_ts_out, VersionUpgradePolicy upgrade_policy,
                std::optional<std::chrono::seconds> heartbeat_interval,
                std::size_t buffer_size, std::string user_agent_ext,
-               databento::Compression compression);
+               databento::Compression compression,
+               std::optional<databento::SlowReadBehavior> slow_read_behavior);
   LiveThreaded(ILogReceiver* log_receiver, std::string key, std::string dataset,
                std::string gateway, std::uint16_t port, bool send_ts_out,
                VersionUpgradePolicy upgrade_policy,
                std::optional<std::chrono::seconds> heartbeat_interval,
                std::size_t buffer_size, std::string user_agent_ext,
-               databento::Compression compression);
+               databento::Compression compression,
+               std::optional<databento::SlowReadBehavior> slow_read_behavior);
 
   // unique_ptr to be movable
   std::unique_ptr<Impl> impl_;
diff --git a/src/enums.cpp b/src/enums.cpp
@@ -108,6 +108,20 @@ const char* ToString(DatasetCondition condition) {
   }
 }
 
+const char* ToString(SlowReadBehavior slow_read_behavior) {
+  switch (slow_read_behavior) {
+    case SlowReadBehavior::Warn: {
+      return "warn";
+    }
+    case SlowReadBehavior::Skip: {
+      return "skip";
+    }
+    default: {
+      return "Unknown";
+    }
+  }
+}
+
 const char* ToString(RType r_type) {
   switch (r_type) {
     case RType::Mbp0: {
@@ -847,6 +861,11 @@ std::ostream& operator<<(std::ostream& out, DatasetCondition condition) {
   return out;
 }
 
+std::ostream& operator<<(std::ostream& out, SlowReadBehavior slow_read_behavior) {
+  out << ToString(slow_read_behavior);
+  return out;
+}
+
 std::ostream& operator<<(std::ostream& out, RType r_type) {
   out << ToString(r_type);
   return out;
diff --git a/src/live.cpp b/src/live.cpp
@@ -84,21 +84,26 @@ LiveBuilder& LiveBuilder::SetCompression(Compression compression) {
   return *this;
 }
 
+LiveBuilder& LiveBuilder::SetSlowReadBehavior(SlowReadBehavior slow_read_behavior) {
+  slow_read_behavior_ = slow_read_behavior;
+  return *this;
+}
+
 databento::LiveBlocking LiveBuilder::BuildBlocking() {
   Validate();
   if (gateway_.empty()) {
     return databento::LiveBlocking{log_receiver_,   key_,
                                    dataset_,        send_ts_out_,
                                    upgrade_policy_, heartbeat_interval_,
                                    buffer_size_,    user_agent_ext_,
-                                   compression_};
+                                   compression_,    slow_read_behavior_};
   }
   return databento::LiveBlocking{log_receiver_,   key_,
                                  dataset_,        gateway_,
                                  port_,           send_ts_out_,
                                  upgrade_policy_, heartbeat_interval_,
                                  buffer_size_,    user_agent_ext_,
-                                 compression_};
+                                 compression_,    slow_read_behavior_};
 }
 
 databento::LiveThreaded LiveBuilder::BuildThreaded() {
@@ -108,14 +113,14 @@ databento::LiveThreaded LiveBuilder::BuildThreaded() {
                                    dataset_,        send_ts_out_,
                                    upgrade_policy_, heartbeat_interval_,
                                    buffer_size_,    user_agent_ext_,
-                                   compression_};
+                                   compression_,    slow_read_behavior_};
   }
   return databento::LiveThreaded{log_receiver_,   key_,
                                  dataset_,        gateway_,
                                  port_,           send_ts_out_,
                                  upgrade_policy_, heartbeat_interval_,
                                  buffer_size_,    user_agent_ext_,
-                                 compression_};
+                                 compression_,    slow_read_behavior_};
 }
 
 void LiveBuilder::Validate() {
diff --git a/src/live_blocking.cpp b/src/live_blocking.cpp
@@ -28,12 +28,12 @@ constexpr std::size_t kBucketIdLength = 5;
 
 databento::LiveBuilder LiveBlocking::Builder() { return databento::LiveBuilder{}; }
 
-LiveBlocking::LiveBlocking(ILogReceiver* log_receiver, std::string key,
-                           std::string dataset, bool send_ts_out,
-                           VersionUpgradePolicy upgrade_policy,
-                           std::optional<std::chrono::seconds> heartbeat_interval,
-                           std::size_t buffer_size, std::string user_agent_ext,
-                           databento::Compression compression)
+LiveBlocking::LiveBlocking(
+    ILogReceiver* log_receiver, std::string key, std::string dataset, bool send_ts_out,
+    VersionUpgradePolicy upgrade_policy,
+    std::optional<std::chrono::seconds> heartbeat_interval, std::size_t buffer_size,
+    std::string user_agent_ext, databento::Compression compression,
+    std::optional<databento::SlowReadBehavior> slow_read_behavior)
 
     : log_receiver_{log_receiver},
       key_{std::move(key)},
@@ -45,16 +45,18 @@ LiveBlocking::LiveBlocking(ILogReceiver* log_receiver, std::string key,
       upgrade_policy_{upgrade_policy},
       heartbeat_interval_{heartbeat_interval},
       compression_{compression},
+      slow_read_behavior_{slow_read_behavior},
       connection_{gateway_, port_},
       buffer_{buffer_size},
       session_id_{this->Authenticate()} {}
 
-LiveBlocking::LiveBlocking(ILogReceiver* log_receiver, std::string key,
-                           std::string dataset, std::string gateway, std::uint16_t port,
-                           bool send_ts_out, VersionUpgradePolicy upgrade_policy,
-                           std::optional<std::chrono::seconds> heartbeat_interval,
-                           std::size_t buffer_size, std::string user_agent_ext,
-                           databento::Compression compression)
+LiveBlocking::LiveBlocking(
+    ILogReceiver* log_receiver, std::string key, std::string dataset,
+    std::string gateway, std::uint16_t port, bool send_ts_out,
+    VersionUpgradePolicy upgrade_policy,
+    std::optional<std::chrono::seconds> heartbeat_interval, std::size_t buffer_size,
+    std::string user_agent_ext, databento::Compression compression,
+    std::optional<databento::SlowReadBehavior> slow_read_behavior)
     : log_receiver_{log_receiver},
       key_{std::move(key)},
       dataset_{std::move(dataset)},
@@ -65,6 +67,7 @@ LiveBlocking::LiveBlocking(ILogReceiver* log_receiver, std::string key,
       upgrade_policy_{upgrade_policy},
       heartbeat_interval_{heartbeat_interval},
       compression_{compression},
+      slow_read_behavior_{slow_read_behavior},
       connection_{gateway_, port_},
       buffer_{buffer_size},
       session_id_{this->Authenticate()} {}
@@ -340,6 +343,9 @@ std::string LiveBlocking::EncodeAuthReq(std::string_view auth) {
   if (heartbeat_interval_.has_value()) {
     req_stream << "|heartbeat_interval_s=" << heartbeat_interval_->count();
   }
+  if (slow_read_behavior_.has_value()) {
+    req_stream << "|slow_read_behavior=" << *slow_read_behavior_;
+  }
   req_stream << '\n';
   return req_stream.str();
 }
diff --git a/src/live_threaded.cpp b/src/live_threaded.cpp
@@ -57,26 +57,28 @@ LiveThreaded::~LiveThreaded() {
   }
 }
 
-LiveThreaded::LiveThreaded(ILogReceiver* log_receiver, std::string key,
-                           std::string dataset, bool send_ts_out,
-                           VersionUpgradePolicy upgrade_policy,
-                           std::optional<std::chrono::seconds> heartbeat_interval,
-                           std::size_t buffer_size, std::string user_agent_ext,
-                           databento::Compression compression)
-    : impl_{std::make_unique<Impl>(
-          log_receiver, std::move(key), std::move(dataset), send_ts_out, upgrade_policy,
-          heartbeat_interval, buffer_size, std::move(user_agent_ext), compression)} {}
-
-LiveThreaded::LiveThreaded(ILogReceiver* log_receiver, std::string key,
-                           std::string dataset, std::string gateway, std::uint16_t port,
-                           bool send_ts_out, VersionUpgradePolicy upgrade_policy,
-                           std::optional<std::chrono::seconds> heartbeat_interval,
-                           std::size_t buffer_size, std::string user_agent_ext,
-                           databento::Compression compression)
+LiveThreaded::LiveThreaded(
+    ILogReceiver* log_receiver, std::string key, std::string dataset, bool send_ts_out,
+    VersionUpgradePolicy upgrade_policy,
+    std::optional<std::chrono::seconds> heartbeat_interval, std::size_t buffer_size,
+    std::string user_agent_ext, databento::Compression compression,
+    std::optional<databento::SlowReadBehavior> slow_read_behavior)
     : impl_{std::make_unique<Impl>(log_receiver, std::move(key), std::move(dataset),
-                                   std::move(gateway), port, send_ts_out,
-                                   upgrade_policy, heartbeat_interval, buffer_size,
-                                   std::move(user_agent_ext), compression)} {}
+                                   send_ts_out, upgrade_policy, heartbeat_interval,
+                                   buffer_size, std::move(user_agent_ext), compression,
+                                   slow_read_behavior)} {}
+
+LiveThreaded::LiveThreaded(
+    ILogReceiver* log_receiver, std::string key, std::string dataset,
+    std::string gateway, std::uint16_t port, bool send_ts_out,
+    VersionUpgradePolicy upgrade_policy,
+    std::optional<std::chrono::seconds> heartbeat_interval, std::size_t buffer_size,
+    std::string user_agent_ext, databento::Compression compression,
+    std::optional<databento::SlowReadBehavior> slow_read_behavior)
+    : impl_{std::make_unique<Impl>(
+          log_receiver, std::move(key), std::move(dataset), std::move(gateway), port,
+          send_ts_out, upgrade_policy, heartbeat_interval, buffer_size,
+          std::move(user_agent_ext), compression, slow_read_behavior)} {}
 
 const std::string& LiveThreaded::Key() const { return impl_->blocking.Key(); }
 
@@ -100,6 +102,10 @@ databento::Compression LiveThreaded::Compression() const {
   return impl_->blocking.Compression();
 }
 
+std::optional<databento::SlowReadBehavior> LiveThreaded::SlowReadBehavior() const {
+  return impl_->blocking.SlowReadBehavior();
+}
+
 const std::vector<databento::LiveSubscription>& LiveThreaded::Subscriptions() const {
   return impl_->blocking.Subscriptions();
 }
diff --git a/tests/include/mock/mock_lsg_server.hpp b/tests/include/mock/mock_lsg_server.hpp
@@ -49,6 +49,8 @@ class MockLsgServer {
                 std::function<void(MockLsgServer&)> serve_fn);
   MockLsgServer(std::string dataset, bool ts_out, Compression compression,
                 std::function<void(MockLsgServer&)> serve_fn);
+  MockLsgServer(std::string dataset, bool ts_out, SlowReadBehavior slow_read_behavior,
+                std::function<void(MockLsgServer&)> serve_fn);
 
   std::uint16_t Port() const { return port_; }
 
@@ -119,6 +121,7 @@ class MockLsgServer {
   bool ts_out_;
   std::chrono::seconds heartbeat_interval_;
   Compression compression_{Compression::None};
+  std::optional<SlowReadBehavior> slow_read_behavior_{};
   std::uint16_t port_{};
   detail::ScopedFd socket_{};
   detail::ScopedFd conn_fd_{};
diff --git a/tests/src/live_blocking_tests.cpp b/tests/src/live_blocking_tests.cpp
@@ -56,6 +56,21 @@ TEST_F(LiveBlockingTests, TestAuthentication) {
                                   .BuildBlocking();
 }
 
+TEST_F(LiveBlockingTests, TestAuthenticationWithSlowReadBehavior) {
+  constexpr auto kTsOut = false;
+  constexpr auto kSlowReadBehavior = SlowReadBehavior::Warn;
+  const mock::MockLsgServer mock_server{dataset::kXnasItch, kTsOut, kSlowReadBehavior,
+                                        [](mock::MockLsgServer& self) {
+                                          self.Accept();
+                                          self.Authenticate();
+                                        }};
+
+  const LiveBlocking target = builder_.SetDataset(dataset::kXnasItch)
+                                  .SetSlowReadBehavior(kSlowReadBehavior)
+                                  .SetAddress(kLocalhost, mock_server.Port())
+                                  .BuildBlocking();
+}
+
 TEST_F(LiveBlockingTests, TestStartAndUpgrade) {
   constexpr auto kTsOut = true;
   for (const auto [upgrade_policy, exp_version] :
diff --git a/tests/src/mock_lsg_server.cpp b/tests/src/mock_lsg_server.cpp
