ct: dl_stm mvcc snapshot

src/v/cloud_topics/BUILD

@@ -41,13 +41,6 @@ redpanda_cc_library(
     ],
 )
 
-redpanda_cc_library(
-    name = "dl_version",
-    hdrs = ["dl_version.h"],
-    include_prefix = "cloud_topics",
-    deps = ["//src/v/utils:named_type"],
-)
-
 redpanda_cc_library(
     name = "dl_overlay",
     srcs = [
@@ -82,6 +75,26 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "dl_version",
+    hdrs = ["dl_version.h"],
+    include_prefix = "cloud_topics",
+    deps = ["//src/v/utils:named_type"],
+)
+
+redpanda_cc_library(
+    name = "dl_snapshot",
+    hdrs = [
+        "dl_snapshot.h",
+    ],
+    include_prefix = "cloud_topics",
+    deps = [
+        ":dl_overlay",
+        ":dl_version",
+        "//src/v/serde",
+    ],
+)
+
 redpanda_cc_library(
     name = "app",
     srcs = [

src/v/cloud_topics/dl_snapshot.h

@@ -0,0 +1,47 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+#pragma once
+
+#include "cloud_topics/dl_overlay.h"
+#include "cloud_topics/dl_version.h"
+#include "container/fragmented_vector.h"
+#include "serde/envelope.h"
+
+namespace experimental::cloud_topics {
+
+struct dl_snapshot_id
+  : serde::
+      envelope<dl_snapshot_id, serde::version<0>, serde::compat_version<0>> {
+    dl_snapshot_id() noexcept = default;
+
+    explicit dl_snapshot_id(dl_version version) noexcept
+      : version(version) {}
+
+    auto serde_fields() { return std::tie(version); }
+
+    bool operator==(const dl_snapshot_id& other) const noexcept = default;
+
+    /// Version for which the snapshot is created.
+    dl_version version;
+};
+
+struct dl_snapshot_payload
+  : serde::checksum_envelope<
+      dl_snapshot_id,
+      serde::version<0>,
+      serde::compat_version<0>> {
+    /// Version for which the snapshot is created.
+    dl_snapshot_id id;
+
+    /// Overlays visible at the snapshot version.
+    fragmented_vector<dl_overlay> overlays;
+};
+
+}; // namespace experimental::cloud_topics

src/v/cloud_topics/dl_stm/BUILD

@@ -7,6 +7,7 @@ redpanda_cc_library(
     visibility = [":__subpackages__"],
     deps = [
         "//src/v/cloud_topics:dl_overlay",
+        "//src/v/cloud_topics:dl_version",
         "//src/v/serde",
     ],
 )
@@ -19,6 +20,7 @@ redpanda_cc_library(
     visibility = [":__subpackages__"],
     deps = [
         "//src/v/cloud_topics:dl_overlay",
+        "//src/v/cloud_topics:dl_snapshot",
         "//src/v/cloud_topics:dl_version",
         "//src/v/container:fragmented_vector",
         "//src/v/model",
@@ -58,6 +60,8 @@ redpanda_cc_library(
     deps = [
         "//src/v/base",
         "//src/v/cloud_topics:dl_overlay",
+        "//src/v/cloud_topics:dl_snapshot",
+        "//src/v/cloud_topics:dl_version",
         "//src/v/model",
         "@seastar",
     ],

src/v/cloud_topics/dl_stm/dl_stm.cc

@@ -46,6 +46,17 @@ ss::future<> dl_stm::do_apply(const model::record_batch& batch) {
             _state.push_overlay(new_dl_version, std::move(cmd.overlay));
             break;
         }
+        case dl_stm_key::start_snapshot: {
+            std::ignore = serde::from_iobuf<start_snapshot_cmd>(
+              r.release_value());
+            _state.start_snapshot(new_dl_version);
+            break;
+        }
+        case dl_stm_key::remove_snapshots_before_version:
+            auto cmd = serde::from_iobuf<remove_snapshots_before_version_cmd>(
+              r.release_value());
+            _state.remove_snapshots_before(cmd.last_version_to_keep);
+            break;
         }
     });
 

src/v/cloud_topics/dl_stm/dl_stm_api.cc

@@ -17,6 +17,8 @@
 #include "serde/rw/uuid.h"
 #include "storage/record_batch_builder.h"
 
+#include <stdexcept>
+
 namespace experimental::cloud_topics {
 
 std::ostream& operator<<(std::ostream& o, dl_stm_api_errc errc) {
@@ -68,4 +70,81 @@ std::optional<dl_overlay> dl_stm_api::lower_bound(kafka::offset offset) const {
     return _stm->_state.lower_bound(offset);
 }
 
+ss::future<checked<dl_snapshot_id, dl_stm_api_errc>>
+dl_stm_api::start_snapshot() {
+    model::term_id term = _stm->_raft->term();
+
+    vlog(_logger.debug, "Replicating dl_stm_cmd::start_snapshot_cmd");
+
+    storage::record_batch_builder builder(
+      model::record_batch_type::dl_stm_command, model::offset(0));
+    builder.add_raw_kv(
+      serde::to_iobuf(dl_stm_key::start_snapshot),
+      serde::to_iobuf(start_snapshot_cmd()));
+
+    auto batch = std::move(builder).build();
+    auto reader = model::make_memory_record_batch_reader(std::move(batch));
+
+    auto opts = raft::replicate_options(raft::consistency_level::quorum_ack);
+    opts.set_force_flush();
+    auto res = co_await _stm->_raft->replicate(term, std::move(reader), opts);
+
+    if (res.has_error()) {
+        throw std::runtime_error(
+          fmt::format("Failed to replicate snapshot: {}", res.error()));
+    }
+
+    // We abuse knowledge of implementation detail here to construct the
+    // dl_snapshot_id without having to setup listeners and notifiers of command
+    // apply.
+    auto expected_id = dl_snapshot_id(dl_version(res.value().last_offset));
+
+    auto applied = co_await _stm->wait_no_throw(
+      res.value().last_offset, model::timeout_clock::now() + 30s);
+    if (!applied) {
+        co_return outcome::failure(dl_stm_api_errc::timeout);
+    }
+
+    // Ensure that the expected snapshot was created.
+    vassert(_stm->_state.snapshot_exists(expected_id), "Snapshot not found");
+
+    co_return outcome::success(expected_id);
+}
+
+std::optional<dl_snapshot_payload>
+dl_stm_api::read_snapshot(dl_snapshot_id id) {
+    return _stm->_state.read_snapshot(id);
+}
+
+ss::future<checked<void, dl_stm_api_errc>>
+dl_stm_api::remove_snapshots_before(dl_version last_version_to_keep) {
+    model::term_id term = _stm->_raft->term();
+
+    vlog(_logger.debug, "Replicating dl_stm_cmd::remove_snapshots_cmd");
+
+    storage::record_batch_builder builder(
+      model::record_batch_type::dl_stm_command, model::offset(0));
+    builder.add_raw_kv(
+      serde::to_iobuf(dl_stm_key::remove_snapshots_before_version),
+      serde::to_iobuf(
+        remove_snapshots_before_version_cmd(last_version_to_keep)));
+
+    auto batch = std::move(builder).build();
+    auto reader = model::make_memory_record_batch_reader(std::move(batch));
+
+    auto opts = raft::replicate_options(raft::consistency_level::quorum_ack);
+    opts.set_force_flush();
+    auto res = co_await _stm->_raft->replicate(term, std::move(reader), opts);
+
+    if (res.has_error()) {
+        throw std::runtime_error(
+          fmt::format("Failed to replicate remove snapshots: {}", res.error()));
+    }
+
+    co_await _stm->wait_no_throw(
+      res.value().last_offset, model::timeout_clock::now() + 30s);
+
+    co_return outcome::success();
+}
+
 }; // namespace experimental::cloud_topics

src/v/cloud_topics/dl_stm/dl_stm_api.h

@@ -11,6 +11,8 @@
 
 #include "base/outcome.h"
 #include "cloud_topics/dl_overlay.h"
+#include "cloud_topics/dl_snapshot.h"
+#include "cloud_topics/dl_version.h"
 
 #include <seastar/util/log.hh>
 
@@ -41,6 +43,18 @@ class dl_stm_api {
     /// available offset.
     std::optional<dl_overlay> lower_bound(kafka::offset offset) const;
 
+    /// Request a new snapshot to be created.
+    ss::future<checked<dl_snapshot_id, dl_stm_api_errc>> start_snapshot();
+
+    /// Read the payload of a snapshot.
+    std::optional<dl_snapshot_payload> read_snapshot(dl_snapshot_id id);
+
+    /// Remove all snapshots with version less than the given version.
+    /// This must be called periodically as new snapshots are being created
+    /// to avoid the state growing indefinitely.
+    ss::future<checked<void, dl_stm_api_errc>>
+    remove_snapshots_before(dl_version last_version_to_keep);
+
 private:
     ss::logger& _logger;
 

src/v/cloud_topics/dl_stm/dl_stm_commands.h

@@ -10,6 +10,7 @@
 #pragma once
 
 #include "cloud_topics/dl_overlay.h"
+#include "cloud_topics/dl_version.h"
 #include "model/fundamental.h"
 #include "model/timestamp.h"
 #include "serde/envelope.h"
@@ -28,4 +29,29 @@ struct push_overlay_cmd
     dl_overlay overlay;
 };
 
+struct start_snapshot_cmd
+  : public serde::envelope<
+      start_snapshot_cmd,
+      serde::version<0>,
+      serde::compat_version<0>> {
+    start_snapshot_cmd() noexcept = default;
+
+    auto serde_fields() { return std::tie(); }
+};
+
+struct remove_snapshots_before_version_cmd
+  : public serde::envelope<
+      remove_snapshots_before_version_cmd,
+      serde::version<0>,
+      serde::compat_version<0>> {
+    remove_snapshots_before_version_cmd() noexcept = default;
+    explicit remove_snapshots_before_version_cmd(
+      dl_version last_version_to_keep)
+      : last_version_to_keep(last_version_to_keep) {}
+
+    auto serde_fields() { return std::tie(last_version_to_keep); }
+
+    dl_version last_version_to_keep{};
+};
+
 } // namespace experimental::cloud_topics

src/v/cloud_topics/dl_stm/dl_stm_state.cc

@@ -10,6 +10,7 @@
 #include "cloud_topics/dl_stm/dl_stm_state.h"
 
 #include "cloud_topics/dl_overlay.h"
+#include "cloud_topics/dl_snapshot.h"
 #include "model/fundamental.h"
 
 #include <algorithm>
@@ -71,4 +72,62 @@ dl_stm_state::lower_bound(kafka::offset offset) const {
     return best_match;
 }
 
+dl_snapshot_id dl_stm_state::start_snapshot(dl_version version) noexcept {
+    _version_invariant.set_last_snapshot_version(version);
+
+    auto id = dl_snapshot_id(version);
+    _snapshots.push_back(id);
+
+    return id;
+}
+
+bool dl_stm_state::snapshot_exists(dl_snapshot_id id) const noexcept {
+    return std::binary_search(
+      _snapshots.begin(),
+      _snapshots.end(),
+      id,
+      [](const dl_snapshot_id& a, const dl_snapshot_id& b) {
+          return a.version < b.version;
+      });
+}
+
+std::optional<dl_snapshot_payload>
+dl_stm_state::read_snapshot(dl_snapshot_id id) const {
+    auto it = std::find_if(
+      _snapshots.begin(), _snapshots.end(), [&id](const dl_snapshot_id& s) {
+          return s.version == id.version;
+      });
+
+    // Snapshot not found.
+    if (it == _snapshots.end()) {
+        return std::nullopt;
+    }
+
+    // Collect overlays that are visible at the snapshot version.
+    fragmented_vector<dl_overlay> overlays;
+    for (const auto& entry : _overlays) {
+        if (
+          entry.added_at <= id.version
+          && (entry.removed_at == dl_version{} || entry.removed_at > id.version)) {
+            overlays.push_back(entry.overlay);
+        }
+    }
+
+    return dl_snapshot_payload{
+      .id = *it,
+      .overlays = std::move(overlays),
+    };
+}
+
+void dl_stm_state::remove_snapshots_before(dl_version last_version_to_keep) {
+    auto it = std::remove_if(
+      _snapshots.begin(),
+      _snapshots.end(),
+      [last_version_to_keep](const dl_snapshot_id& id) {
+          return id.version < last_version_to_keep;
+      });
+
+    _snapshots.erase(it, _snapshots.end());
+}
+
 } // namespace experimental::cloud_topics