PS-10246 feature: Implement fixing next_event_position fields / checksums in the event common_header / footer in case of reconnecting to another node (part 3)

src/binsrv/ctime_timestamp.cpp

@@ -15,6 +15,7 @@
 
 #include "binsrv/ctime_timestamp.hpp"
 
+#include <ctime>
 #include <exception>
 #include <ios>
 #include <istream>
@@ -45,14 +46,23 @@ ctime_timestamp::try_parse(std::string_view value_sv,
   return result;
 }
 
-[[nodiscard]] std::string ctime_timestamp::str() const {
+[[nodiscard]] std::string ctime_timestamp::iso_extended_str() const {
   return boost::posix_time::to_iso_extended_string(
       boost::posix_time::from_time_t(get_value()));
 }
 
+[[nodiscard]] std::string ctime_timestamp::simple_str() const {
+  return boost::posix_time::to_simple_string(
+      boost::posix_time::from_time_t(get_value()));
+}
+
+[[nodiscard]] ctime_timestamp ctime_timestamp::now() noexcept {
+  return ctime_timestamp{std::time(nullptr)};
+}
+
 std::ostream &operator<<(std::ostream &output,
                          const ctime_timestamp &timestamp) {
-  return output << timestamp.str();
+  return output << timestamp.iso_extended_str();
 }
 
 std::istream &operator>>(std::istream &input, ctime_timestamp &timestamp) {

src/binsrv/ctime_timestamp.hpp

@@ -40,7 +40,10 @@ class [[nodiscard]] ctime_timestamp {
   void reset_to_epoch() noexcept { value_ = std::time_t{}; }
 
   [[nodiscard]] std::time_t get_value() const noexcept { return value_; }
-  [[nodiscard]] std::string str() const;
+  [[nodiscard]] std::string iso_extended_str() const;
+  [[nodiscard]] std::string simple_str() const;
+
+  [[nodiscard]] static ctime_timestamp now() noexcept;
 
   friend auto operator<=>(const ctime_timestamp &first,
                           const ctime_timestamp &second) = default;

src/binsrv/events/checksum_algorithm_type.hpp

@@ -54,7 +54,7 @@ inline std::string_view to_string_view(checksum_algorithm_type code) noexcept {
 #undef BINSRV_CHECKSUM_ALGORITHM_TYPE_XY_MACRO
   // NOLINTNEXTLINE(llvm-qualified-auto,readability-qualified-auto)
   const auto fnd{std::ranges::find(labels, code, &nv_pair::first)};
-  return fnd == std::end(labels) ? ""sv : fnd->second;
+  return fnd == std::cend(labels) ? ""sv : fnd->second;
 }
 #undef BINSRV_CHECKSUM_ALGORITHM_TYPE_XY_SEQUENCE
 // NOLINTEND(cppcoreguidelines-macro-usage)

src/binsrv/events/code_type.hpp

@@ -94,7 +94,7 @@ inline std::string_view to_string_view(code_type code) noexcept {
 #undef BINSRV_EVENTS_CODE_TYPE_XY_MACRO
   // NOLINTNEXTLINE(llvm-qualified-auto,readability-qualified-auto)
   const auto fnd{std::ranges::find(labels, code, &nv_pair::first)};
-  return fnd == std::end(labels) ? ""sv : fnd->second;
+  return fnd == std::cend(labels) ? ""sv : fnd->second;
 }
 #undef BINSRV_EVENTS_CODE_TYPE_XY_SEQUENCE
 // NOLINTEND(cppcoreguidelines-macro-usage)

src/binsrv/events/common_header.cpp

@@ -15,19 +15,35 @@
 
 #include "binsrv/events/common_header.hpp"
 
+#include <cstdint>
+#include <iterator>
 #include <ostream>
+#include <stdexcept>
 
 #include <boost/align/align_up.hpp>
 
 #include "binsrv/ctime_timestamp.hpp"
 
+#include "binsrv/events/code_type.hpp"
 #include "binsrv/events/common_header_flag_type.hpp"
 #include "binsrv/events/common_header_view.hpp"
 
 #include "util/byte_span_fwd.hpp"
+#include "util/conversion_helpers.hpp"
+#include "util/exception_location_helpers.hpp"
 
 namespace binsrv::events {
 
+common_header::common_header(
+    const ctime_timestamp &timestamp, code_type type_code,
+    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
+    std::uint32_t server_id, std::uint32_t event_size,
+    std::uint32_t next_event_position, common_header_flag_set flags) noexcept
+    : timestamp_(static_cast<std::uint32_t>(timestamp.get_value())),
+      server_id_(server_id), event_size_(event_size),
+      next_event_position_(next_event_position), flags_(flags.get_bits()),
+      type_code_(util::to_underlying(type_code)) {}
+
 common_header::common_header(const common_header_view &view)
     : timestamp_{view.get_timestamp_raw()},
       server_id_{view.get_server_id_raw()},
@@ -50,6 +66,20 @@ common_header::common_header(const common_header_view &view)
 common_header::common_header(util::const_byte_span portion)
     : common_header{common_header_view{portion}} {}
 
+[[nodiscard]] common_header common_header::create_with_offset(
+    std::uint32_t offset, std::uint32_t event_size,
+    const ctime_timestamp &timestamp, code_type type_code,
+    std::uint32_t server_id, common_header_flag_set flags) noexcept {
+  // artificial ROTATE event must have next_event_position set to zero
+  const std::uint32_t next_event_position{
+      type_code == code_type::rotate &&
+              flags.has_element(common_header_flag_type::artificial)
+          ? 0U
+          : offset + event_size};
+  return common_header{timestamp,  type_code,           server_id,
+                       event_size, next_event_position, flags};
+}
+
 [[nodiscard]] ctime_timestamp common_header::get_timestamp() const noexcept {
   return common_header_view_base::get_timestamp_from_raw(get_timestamp_raw());
 }
@@ -58,6 +88,22 @@ common_header::common_header(util::const_byte_span portion)
   return common_header_view_base::get_flags_from_raw(get_flags_raw());
 }
 
+void common_header::encode_to(util::byte_span &destination) const {
+  if (std::size(destination) < calculate_encoded_size()) {
+    util::exception_location().raise<std::invalid_argument>(
+        "cannot encode common header");
+  }
+  const common_header_updatable_view common_header_uv{
+      destination.subspan(0U, calculate_encoded_size())};
+  common_header_uv.set_timestamp_raw(get_timestamp_raw());
+  common_header_uv.set_type_code_raw(get_type_code_raw());
+  common_header_uv.set_server_id_raw(get_server_id_raw());
+  common_header_uv.set_event_size_raw(get_event_size_raw());
+  common_header_uv.set_next_event_position_raw(get_next_event_position_raw());
+  common_header_uv.set_flags_raw(get_flags_raw());
+  destination = destination.subspan(calculate_encoded_size());
+}
+
 std::ostream &operator<<(std::ostream &output, const common_header &obj) {
   return output << "ts: " << obj.get_readable_timestamp()
                 << ", type: " << obj.get_readable_type_code()

src/binsrv/events/common_header.hpp

@@ -39,9 +39,19 @@ class [[nodiscard]] common_header {
   static constexpr std::size_t size_in_bytes{
       common_header_view_base::size_in_bytes};
 
+  common_header(const ctime_timestamp &timestamp, code_type type_code,
+                std::uint32_t server_id, std::uint32_t event_size,
+                std::uint32_t next_event_position,
+                common_header_flag_set flags) noexcept;
   explicit common_header(const common_header_view &view);
   explicit common_header(util::const_byte_span portion);
 
+  [[nodiscard]] static common_header
+  create_with_offset(std::uint32_t offset, std::uint32_t event_size,
+                     const ctime_timestamp &timestamp, code_type type_code,
+                     std::uint32_t server_id,
+                     common_header_flag_set flags) noexcept;
+
   [[nodiscard]] std::uint32_t get_timestamp_raw() const noexcept {
     return timestamp_;
   }
@@ -81,6 +91,14 @@ class [[nodiscard]] common_header {
         get_flags_raw());
   }
 
+  [[nodiscard]] static std::size_t calculate_encoded_size() noexcept {
+    return size_in_bytes;
+  }
+  void encode_to(util::byte_span &destination) const;
+
+  friend bool operator==(const common_header &first,
+                         const common_header &second) = default;
+
 private:
   // the members are deliberately reordered for better packing
   std::uint32_t timestamp_{};           // 0

src/binsrv/events/common_header_flag_type.hpp

@@ -70,7 +70,7 @@ inline std::string_view to_string_view(common_header_flag_type code) noexcept {
 #undef BINSRV_EVENTS_COMMON_HEADER_FLAG_TYPE_XY_MACRO
   // NOLINTNEXTLINE(llvm-qualified-auto,readability-qualified-auto)
   const auto fnd{std::ranges::find(labels, code, &nv_pair::first)};
-  return fnd == std::end(labels) ? ""sv : fnd->second;
+  return fnd == std::cend(labels) ? ""sv : fnd->second;
 }
 #undef BINSRV_EVENTS_COMMON_HEADER_FLAG_TYPE_XY_SEQUENCE
 // NOLINTEND(cppcoreguidelines-macro-usage)

src/binsrv/events/common_header_view.cpp

@@ -44,7 +44,7 @@ namespace binsrv::events {
 [[nodiscard]] std::string
 common_header_view_base::get_readable_timestamp_from_raw(
     std::uint32_t timestamp) {
-  return get_timestamp_from_raw(timestamp).str();
+  return get_timestamp_from_raw(timestamp).iso_extended_str();
 }
 
 [[nodiscard]] code_type common_header_view_base::get_type_code_from_raw(

src/binsrv/events/empty_body.hpp

@@ -26,7 +26,16 @@ class [[nodiscard]] empty_body {
 public:
   static constexpr std::size_t size_in_bytes{0U};
 
+  empty_body() = default;
   explicit empty_body(util::const_byte_span portion);
+
+  [[nodiscard]] static std::size_t calculate_encoded_size() noexcept {
+    return size_in_bytes;
+  }
+  void encode_to(util::byte_span & /* destination */) const noexcept {}
+
+  friend bool operator==(const empty_body & /* first */,
+                         const empty_body & /* second */) = default;
 };
 
 } // namespace binsrv::events

src/binsrv/events/empty_post_header.hpp

@@ -26,7 +26,16 @@ class [[nodiscard]] empty_post_header {
 public:
   static constexpr std::size_t size_in_bytes{0U};
 
+  empty_post_header() = default;
   explicit empty_post_header(util::const_byte_span portion);
+
+  [[nodiscard]] static std::size_t calculate_encoded_size() noexcept {
+    return size_in_bytes;
+  }
+  void encode_to(util::byte_span & /* destination */) const noexcept {}
+
+  friend bool operator==(const empty_post_header & /* first */,
+                         const empty_post_header & /* second */) = default;
 };
 
 } // namespace binsrv::events

src/binsrv/events/event.cpp

@@ -17,9 +17,11 @@
 
 #include <array>
 #include <cassert>
+#include <concepts>
 #include <cstddef>
 #include <cstdint>
 #include <ostream>
+#include <stdexcept>
 #include <utility>
 #include <variant>
 
@@ -32,6 +34,8 @@
 
 #include "util/byte_span_fwd.hpp"
 #include "util/conversion_helpers.hpp"
+#include "util/crc_helpers.hpp"
+#include "util/exception_location_helpers.hpp"
 
 namespace binsrv::events {
 
@@ -53,6 +57,49 @@ event::event(reader_context &context, const event_view &view)
 event::event(reader_context &context, util::const_byte_span portion)
     : event{context, event_view{context, portion}} {}
 
+template <typename T>
+concept encodable = requires(const T &obj, util::byte_span &destination) {
+  { obj.calculate_encoded_size() } -> std::same_as<std::size_t>;
+  { obj.encode_to(destination) };
+};
+
+[[nodiscard]] std::size_t event::calculate_encoded_size() const {
+  const auto size_calculation_visitor =
+      [](const auto &component) -> std::size_t {
+    if constexpr (encodable<decltype(component)>) {
+      return component.calculate_encoded_size();
+    } else {
+      util::exception_location().raise<std::logic_error>(
+          "calculate_encoded_size() not implemented for this event post header "
+          "/ body");
+    }
+  };
+  return common_header::calculate_encoded_size() +
+         std::visit(size_calculation_visitor, post_header_) +
+         std::visit(size_calculation_visitor, body_) +
+         (footer_ ? footer::calculate_encoded_size() : 0U);
+}
+void event::encode_to(util::byte_span &destination) const {
+  if (std::size(destination) < calculate_encoded_size()) {
+    util::exception_location().raise<std::invalid_argument>(
+        "cannot encode event");
+  }
+  const auto encoding_visitor = [&destination](const auto &component) {
+    if constexpr (encodable<decltype(component)>) {
+      component.encode_to(destination);
+    } else {
+      util::exception_location().raise<std::logic_error>(
+          "encode_to() not implemented for this event post header / body");
+    }
+  };
+  common_header_.encode_to(destination);
+  std::visit(encoding_visitor, post_header_);
+  std::visit(encoding_visitor, body_);
+  if (footer_) {
+    footer_->encode_to(destination);
+  }
+}
+
 void event::emplace_post_header(std::uint32_t encoded_server_version,
                                 code_type code, util::const_byte_span portion) {
   // our goal here is to initialize (emplace) a specific class inside
@@ -125,6 +172,25 @@ void event::emplace_body(std::uint32_t encoded_server_version, code_type code,
   (this->*emplace_functions[function_index])(encoded_server_version, portion);
 }
 
+void event::encode_and_checksum(event_storage &buffer, bool include_checksum) {
+  const auto size_with_footer{common_header_.get_event_size_raw()};
+  buffer.resize(size_with_footer);
+  util::byte_span output_span(std::data(buffer), size_with_footer);
+  // encoding event to the buffer
+  encode_to(output_span);
+  if (include_checksum) {
+    // calculating checksum
+    const auto size_wo_footer{size_with_footer -
+                              footer::calculate_encoded_size()};
+    const util::const_byte_span crc_span{std::data(buffer), size_wo_footer};
+    const auto crc{util::calculate_crc32(crc_span)};
+    footer_.emplace(crc);
+    // updating crc in the footer zone of the buffer
+    const footer_updatable_view footer_uv{output_span};
+    footer_uv.set_crc_raw(crc);
+  }
+}
+
 std::ostream &operator<<(std::ostream &output, const event &obj) {
   output << "| common header | " << obj.get_common_header() << " |";
 
@@ -136,9 +202,9 @@ std::ostream &operator<<(std::ostream &output, const event &obj) {
   output << " |\n| body          | ";
   std::visit(generic_printer, obj.get_generic_body());
   output << " |";
-  const auto &footer{obj.get_footer()};
-  if (footer) {
-    output << "\n| footer        | " << *footer << " |";
+  const auto &opt_footer{obj.get_footer()};
+  if (opt_footer.has_value()) {
+    output << "\n| footer        | " << *opt_footer << " |";
   }
   return output;
 }