ct/l0: add better logging for stale epochs

Right now we get unhelpful log messages about empty epochs, but now at
least we get a better error message about what the latest value in the
STM is.

Author: rockwotj

src/v/cloud_topics/frontend/frontend.cc

@@ -580,12 +580,13 @@ ss::future<result<raft::replicate_result>> do_upload_and_replicate(
         co_return default_errc;
     }
     auto fence = std::move(fence_fut.get());
-    if (!fence.unit.has_value()) {
+    if (!fence.has_value()) {
         vlog(
           cd_log.warn,
-          "Failed to fence epoch {} for ntp {}, fence unit is empty",
+          "Failed to fence epoch {} for ntp {}, ctp latest seen epoch is {}",
           upload_res.value().front().id.epoch,
-          ntp);
+          ntp,
+          fence.error().latest_seen);
         co_return default_errc;
     }
 
@@ -605,7 +606,7 @@ ss::future<result<raft::replicate_result>> do_upload_and_replicate(
         co_await ticket.redeem();
     }
     // Replicate now that our ticket is redeemed
-    opts = update_replicate_options(opts, fence.term);
+    opts = update_replicate_options(opts, fence->term);
     auto replicate_stages = partition->replicate_in_stages(
       batch_id, std::move(placeholders.batches.front()), opts);
     // Once the request is enqueued in raft and our order is guaranteed we can
@@ -722,14 +723,13 @@ ss::future<std::expected<kafka::offset, std::error_code>> frontend::replicate(
         std::rethrow_exception(e);
     }
     auto fence = std::move(fence_fut.get());
-    if (!fence.unit.has_value()) {
+    if (!fence.has_value()) {
         vlog(
           cd_log.warn,
-          "Failed to fence epoch {} for ntp {}, fence unit is empty",
+          "Failed to fence epoch {} for ntp {}, ctp latest seen epoch is {}",
           res.value().front().id.epoch,
-          ntp());
-
-        /// TODO: Maybe return different error code here?
+          ntp(),
+          fence.error().latest_seen);
         co_return std::unexpected(
           kafka::make_error_code(kafka::error_code::request_timed_out));
     }
@@ -741,7 +741,7 @@ ss::future<std::expected<kafka::offset, std::error_code>> frontend::replicate(
         placeholder_batches.push_back(std::move(batch));
     }
 
-    opts = update_replicate_options(opts, fence.term);
+    opts = update_replicate_options(opts, fence->term);
     auto result = co_await _partition->replicate(
       std::move(placeholder_batches), opts);
 

src/v/cloud_topics/level_zero/stm/BUILD

@@ -63,6 +63,7 @@ redpanda_cc_library(
     hdrs = ["types.h"],
     visibility = [":__subpackages__"],
     deps = [
+        "//src/v/cloud_topics:types",
         "//src/v/model",
         "@fmt",
         "@seastar",

src/v/cloud_topics/level_zero/stm/ctp_stm.cc

@@ -332,7 +332,8 @@ ss::future<iobuf> ctp_stm::take_raft_snapshot(model::offset snapshot_at) {
     co_return serde::to_iobuf(_state);
 }
 
-ss::future<cluster_epoch_fence> ctp_stm::fence_epoch(cluster_epoch e) {
+ss::future<std::expected<cluster_epoch_fence, stale_cluster_epoch>>
+ctp_stm::fence_epoch(cluster_epoch e) {
     auto holder = _gate.hold();
     if (!co_await sync(sync_timeout)) {
         vlog(_log.warn, "ctp_stm::fence_epoch sync timeout");
@@ -366,7 +367,10 @@ ss::future<cluster_epoch_fence> ctp_stm::fence_epoch(cluster_epoch e) {
         }
     }
     // If we reach here, it means that we need to discard the batch.
-    co_return cluster_epoch_fence{};
+    co_return std::unexpected(
+      stale_cluster_epoch(_state.get_max_seen_epoch()
+                            .or_else(get_applied_epoch)
+                            .value_or(cluster_epoch{-1})));
 }
 
 model::offset ctp_stm::max_removable_local_log_offset() {

src/v/cloud_topics/level_zero/stm/ctp_stm.h

@@ -17,6 +17,8 @@
 
 #include <seastar/core/rwlock.hh>
 
+#include <expected>
+
 namespace cloud_topics {
 
 class ctp_stm_api;
@@ -54,7 +56,8 @@ class ctp_stm final : public raft::persisted_stm<> {
         _state.advance_max_seen_epoch(epoch);
     }
 
-    ss::future<cluster_epoch_fence> fence_epoch(cluster_epoch e);
+    ss::future<std::expected<cluster_epoch_fence, stale_cluster_epoch>>
+    fence_epoch(cluster_epoch e);
 
     /// Return inactive epoch of the CTP
     ///

src/v/cloud_topics/level_zero/stm/ctp_stm_api.cc

@@ -174,14 +174,15 @@ ss::future<bool> ctp_stm_api::sync_in_term(
     co_return co_await _stm->sync_in_term(deadline, as);
 }
 
-ss::future<cluster_epoch_fence> ctp_stm_api::fence_epoch(cluster_epoch e) {
+ss::future<std::expected<cluster_epoch_fence, stale_cluster_epoch>>
+ctp_stm_api::fence_epoch(cluster_epoch e) {
     vlog(_log.debug, "Fencing epoch {} in term {}", e, _stm->_raft->term());
     auto res = co_await _stm->fence_epoch(e);
     vlog(
       _log.debug,
       "Fence acquired = {} in term {}",
-      res.unit.has_value(),
-      res.term);
+      res.has_value(),
+      res.has_value() ? res->term : model::term_id{-1});
     co_return std::move(res);
 }
 

src/v/cloud_topics/level_zero/stm/ctp_stm_api.h

@@ -91,7 +91,8 @@ class ctp_stm_api {
     sync_in_term(model::timeout_clock::time_point deadline, ss::abort_source&);
 
     /// Fence writes
-    ss::future<cluster_epoch_fence> fence_epoch(cluster_epoch e);
+    ss::future<std::expected<cluster_epoch_fence, stale_cluster_epoch>>
+    fence_epoch(cluster_epoch e);
 
     std::optional<cluster_epoch> get_max_epoch() const;
 

src/v/cloud_topics/level_zero/stm/tests/ctp_stm_test.cc

@@ -158,20 +158,20 @@ TEST_F_CORO(ctp_stm_fixture, test_fencing) {
     {
         auto fence
           = co_await api(node(*get_leader())).fence_epoch(ct::cluster_epoch{2});
-        ASSERT_TRUE_CORO(fence.unit.has_value());
+        ASSERT_TRUE_CORO(fence.has_value());
     }
 
     // Acquire the fence for epoch 1 (should fail)
     {
         auto fence
           = co_await api(node(*get_leader())).fence_epoch(ct::cluster_epoch{1});
-        ASSERT_FALSE_CORO(fence.unit.has_value());
+        ASSERT_FALSE_CORO(fence.has_value());
     }
 
     // Advance max_seen_epoch to 3.
     auto write_fence
       = co_await api(node(*get_leader())).fence_epoch(ct::cluster_epoch{3});
-    ASSERT_TRUE_CORO(write_fence.unit.has_value());
+    ASSERT_TRUE_CORO(write_fence.has_value());
 
     // Out of order fence for epoch 2 (should be waiting for the fence to be
     // released)
@@ -182,7 +182,7 @@ TEST_F_CORO(ctp_stm_fixture, test_fencing) {
     write_fence = {};
 
     auto read_fence = co_await std::move(fut);
-    ASSERT_FALSE_CORO(read_fence.unit.has_value());
+    ASSERT_FALSE_CORO(read_fence.has_value());
 }
 
 TEST_F_CORO(ctp_stm_fixture, test_last_reconciled_offset) {
@@ -457,6 +457,6 @@ TEST_F_CORO(ctp_stm_fixture, test_snapshot) {
     // Acquire the fence for epoch 1 (should fail)
     {
         auto fence = co_await api(leader).fence_epoch(ct::cluster_epoch{1});
-        ASSERT_FALSE_CORO(fence.unit.has_value());
+        ASSERT_FALSE_CORO(fence.has_value());
     }
 }

src/v/cloud_topics/level_zero/stm/types.h

@@ -10,6 +10,7 @@
 
 #pragma once
 
+#include "cloud_topics/types.h"
 #include "model/fundamental.h"
 
 #include <seastar/core/rwlock.hh>
@@ -23,12 +24,18 @@ enum class ctp_stm_key : uint8_t {
 
 struct [[nodiscard]] cluster_epoch_fence {
     // Units protecting the epoch state.
-    // If it's set to nullopt the batch has to be discarded
-    // because of the out of order epoch.
-    std::optional<ss::rwlock::holder> unit;
+    ss::rwlock::holder unit;
     // Term in which the batch is replicated.
     model::term_id term;
 };
+
+// The error returned when the CTP STM has seen a newer epoch than the one
+// attempting to be used.
+struct [[nodiscard]] stale_cluster_epoch {
+    // The latest cluster epoch
+    cluster_epoch latest_seen;
+};
+
 } // namespace cloud_topics
 
 template<>