ADD: Add live heartbeat monitoring

Author: threecgreen

CHANGELOG.md

@@ -3,8 +3,14 @@
 ## 0.52.0 - TBD
 
 ### Enhancements
+- Added client-side heartbeat timeout detection: `NextRecord` throws `HeartbeatTimeoutError`
+  if no data is received for `heartbeat_interval` + 5 seconds (defaults to 35 seconds)
 - Changed `SlowReaderBehavior::Skip` to send "skip" instead of "drop" to the gateway
 
+### Breaking changes
+- `NextRecord` now throws `LiveApiError` instead of `DbnResponseError` when the gateway
+  closes the session. Code catching `DbnResponseError` for this case should be updated
+
 ## 0.51.0 - 2026-03-17
 
 ### Enhancements

CMakeLists.txt

@@ -202,13 +202,16 @@ if(${PROJECT_NAME_UPPERCASE}_USE_EXTERNAL_DATE)
     find_package(date REQUIRED)
   endif()
 else()
-  set(date_version 3.0.3)
+  set(date_version 3.0.4)
+  set(CMAKE_WARN_DEPRECATED OFF CACHE BOOL "" FORCE)
   FetchContent_Declare(
     date_src
     URL https://github.com/HowardHinnant/date/archive/refs/tags/v${date_version}.tar.gz
     DOWNLOAD_EXTRACT_TIMESTAMP TRUE
   )
+  # Suppress deprecation warning from date's old cmake_minimum_required
   FetchContent_MakeAvailable(date_src)
+  set(CMAKE_WARN_DEPRECATED ON CACHE BOOL "" FORCE)
   # Ignore compiler warnings in headers
   add_system_include_property(date)
 endif()

cmake/Findzstd.cmake

@@ -1,3 +1,17 @@
+# Try config mode first. If the system provides a zstd CMake config (e.g. Homebrew),
+# use it directly so all targets are defined consistently.
+find_package(zstd CONFIG QUIET)
+if(zstd_FOUND)
+  if(NOT TARGET zstd::libzstd)
+    if(TARGET zstd::libzstd_shared)
+      add_library(zstd::libzstd ALIAS zstd::libzstd_shared)
+    elseif(TARGET zstd::libzstd_static)
+      add_library(zstd::libzstd ALIAS zstd::libzstd_static)
+    endif()
+  endif()
+  return()
+endif()
+
 include(FindPackageHandleStandardArgs)
 
 if(WIN32)

include/databento/detail/live_connection.hpp

@@ -13,12 +13,18 @@
 #include "databento/ireadable.hpp"
 #include "databento/iwritable.hpp"
 
+// Forward declare
+namespace databento {
+class ILogReceiver;
+}
+
 namespace databento::detail {
 // Manages the TCP connection to the live gateway with optionally compressed reads for
 // the DBN data.
 class LiveConnection : IWritable {
  public:
-  LiveConnection(const std::string& gateway, std::uint16_t port);
+  LiveConnection(ILogReceiver* log_receiver, const std::string& gateway,
+                 std::uint16_t port);
 
   void WriteAll(std::string_view str);
   void WriteAll(const std::byte* buffer, std::size_t size);

include/databento/detail/tcp_client.hpp

@@ -9,6 +9,11 @@
 #include "databento/detail/scoped_fd.hpp"  // ScopedFd
 #include "databento/ireadable.hpp"
 
+// Forward declare
+namespace databento {
+class ILogReceiver;
+}
+
 namespace databento::detail {
 class TcpClient {
  public:
@@ -17,8 +22,9 @@ class TcpClient {
     std::chrono::seconds max_wait{std::chrono::minutes{1}};
   };
 
-  TcpClient(const std::string& gateway, std::uint16_t port);
-  TcpClient(const std::string& gateway, std::uint16_t port, RetryConf retry_conf);
+  TcpClient(ILogReceiver* log_receiver, const std::string& gateway, std::uint16_t port);
+  TcpClient(ILogReceiver* log_receiver, const std::string& gateway, std::uint16_t port,
+            RetryConf retry_conf);
 
   void WriteAll(std::string_view str);
   void WriteAll(const std::byte* buffer, std::size_t size);
@@ -32,8 +38,8 @@ class TcpClient {
   void Close();
 
  private:
-  static ScopedFd InitSocket(const std::string& gateway, std::uint16_t port,
-                             RetryConf retry_conf);
+  static ScopedFd InitSocket(ILogReceiver* log_receiver, const std::string& gateway,
+                             std::uint16_t port, RetryConf retry_conf);
 
   ScopedFd socket_;
 };

include/databento/exceptions.hpp

@@ -6,6 +6,7 @@
 #include <httplib.h>          // Error
 #include <nlohmann/json.hpp>  // json, parse_error
 
+#include <chrono>
 #include <cstdint>
 #include <exception>
 #include <string>
@@ -132,6 +133,20 @@ class DbnResponseError : public Exception {
   explicit DbnResponseError(std::string message) : Exception{std::move(message)} {}
 };
 
+// Exception indicating no data was received within the heartbeat timeout window.
+class HeartbeatTimeoutError : public Exception {
+ public:
+  explicit HeartbeatTimeoutError(std::chrono::seconds elapsed)
+      : Exception{BuildMessage(elapsed)}, elapsed_{elapsed} {}
+
+  std::chrono::seconds Elapsed() const { return elapsed_; }
+
+ private:
+  static std::string BuildMessage(std::chrono::seconds elapsed);
+
+  const std::chrono::seconds elapsed_;
+};
+
 // Exception indicating something internal to the live API, but unrelated to TCP
 // went wrong.
 class LiveApiError : public Exception {

include/databento/live_blocking.hpp

@@ -1,7 +1,7 @@
 #pragma once
 
 #include <array>
-#include <chrono>  // milliseconds
+#include <chrono>  // milliseconds, steady_clock
 #include <cstddef>
 #include <cstdint>
 #include <optional>
@@ -119,6 +119,8 @@ class LiveBlocking {
                  bool use_snapshot);
   IReadable::Result FillBuffer(std::chrono::milliseconds timeout);
   RecordHeader* BufferRecordHeader();
+  std::chrono::milliseconds HeartbeatTimeout() const;
+  void CheckHeartbeatTimeout() const;
 
   static constexpr std::size_t kMaxStrLen = 24L * 1024;
 
@@ -142,5 +144,7 @@ class LiveBlocking {
   alignas(RecordHeader) std::array<std::byte, kMaxRecordLen> compat_buffer_{};
   std::uint64_t session_id_;
   Record current_record_{nullptr};
+  std::chrono::steady_clock::time_point last_read_time_{
+      std::chrono::steady_clock::now()};
 };
 }  // namespace databento

src/detail/live_connection.cpp

@@ -6,8 +6,9 @@
 
 using databento::detail::LiveConnection;
 
-LiveConnection::LiveConnection(const std::string& gateway, std::uint16_t port)
-    : client_{gateway, port} {}
+LiveConnection::LiveConnection(ILogReceiver* log_receiver, const std::string& gateway,
+                               std::uint16_t port)
+    : client_{log_receiver, gateway, port} {}
 
 void LiveConnection::WriteAll(std::string_view str) { client_.WriteAll(str); }
 

src/detail/tcp_client.cpp

@@ -5,7 +5,7 @@
 #else
 #include <netdb.h>       // addrinfo, gai_strerror, getaddrinfo, freeaddrinfo
 #include <netinet/in.h>  // htons, IPPROTO_TCP
-#include <sys/poll.h>    // pollfd, POLLHUP
+#include <sys/poll.h>    // pollfd
 #include <sys/socket.h>  // AF_INET, connect, recv, send, sockaddr, sockaddr_in, socket, SOCK_STREAM
 #include <unistd.h>  // close, ssize_t
 
@@ -18,6 +18,7 @@
 #include <thread>
 
 #include "databento/exceptions.hpp"  // TcpError
+#include "databento/log.hpp"         // ILogReceiver
 
 using databento::detail::TcpClient;
 using Status = databento::IReadable::Status;
@@ -32,12 +33,13 @@ int GetErrNo() {
 }
 }  // namespace
 
-TcpClient::TcpClient(const std::string& gateway, std::uint16_t port)
-    : TcpClient{gateway, port, {}} {}
+TcpClient::TcpClient(ILogReceiver* log_receiver, const std::string& gateway,
+                     std::uint16_t port)
+    : TcpClient{log_receiver, gateway, port, {}} {}
 
-TcpClient::TcpClient(const std::string& gateway, std::uint16_t port,
-                     RetryConf retry_conf)
-    : socket_{InitSocket(gateway, port, retry_conf)} {}
+TcpClient::TcpClient(ILogReceiver* log_receiver, const std::string& gateway,
+                     std::uint16_t port, RetryConf retry_conf)
+    : socket_{InitSocket(log_receiver, gateway, port, retry_conf)} {}
 
 void TcpClient::WriteAll(std::string_view str) {
   WriteAll(reinterpret_cast<const std::byte*>(str.data()), str.length());
@@ -103,9 +105,12 @@ databento::IReadable::Result TcpClient::ReadSome(std::byte* buffer,
 
 void TcpClient::Close() { socket_.Close(); }
 
-databento::detail::ScopedFd TcpClient::InitSocket(const std::string& gateway,
+databento::detail::ScopedFd TcpClient::InitSocket(ILogReceiver* log_receiver,
+                                                  const std::string& gateway,
                                                   std::uint16_t port,
                                                   RetryConf retry_conf) {
+  static constexpr auto kMethod = "TcpClient::TcpClient";
+
   const detail::Socket fd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
   if (fd == -1) {
     throw TcpError{::GetErrNo(), "Failed to create socket"};
@@ -121,7 +126,7 @@ databento::detail::ScopedFd TcpClient::InitSocket(const std::string& gateway,
   const auto ret =
       ::getaddrinfo(gateway.c_str(), std::to_string(port).c_str(), &hints, &out);
   if (ret != 0) {
-    throw InvalidArgumentError{"TcpClient::TcpClient", "addr", ::gai_strerror(ret)};
+    throw InvalidArgumentError{kMethod, "addr", ::gai_strerror(ret)};
   }
   std::unique_ptr<addrinfo, decltype(&::freeaddrinfo)> res{out, &::freeaddrinfo};
   const auto max_attempts = std::max<std::uint32_t>(retry_conf.max_attempts, 1);
@@ -134,7 +139,12 @@ databento::detail::ScopedFd TcpClient::InitSocket(const std::string& gateway,
       err_msg << "Socket failed to connect after " << max_attempts << " attempts";
       throw TcpError{::GetErrNo(), err_msg.str()};
     }
-    // TODO(cg): Log
+    std::ostringstream log_msg;
+    log_msg << '[' << kMethod << "] Connection attempt " << (attempt + 1) << " to "
+            << gateway << ':' << port << " failed, retrying in " << backoff.count()
+            << " seconds";
+    log_receiver->Receive(LogLevel::Warning, log_msg.str());
+
     std::this_thread::sleep_for(backoff);
     backoff = (std::min)(backoff * 2, retry_conf.max_wait);
   }

src/exceptions.cpp

@@ -89,6 +89,15 @@ JsonResponseError JsonResponseError::TypeMismatch(std::string_view method_name,
   return JsonResponseError{err_msg.str()};
 }
 
+using databento::HeartbeatTimeoutError;
+
+std::string HeartbeatTimeoutError::BuildMessage(std::chrono::seconds elapsed) {
+  std::ostringstream err_msg;
+  err_msg << "Heartbeat timeout: no data received for " << elapsed.count()
+          << " seconds";
+  return err_msg.str();
+}
+
 using databento::LiveApiError;
 
 LiveApiError LiveApiError::UnexpectedMsg(std::string_view message,