ct/fe: frontend::advance_epoch

- fence
- advance epoch thru ctp_stm_api
- sync to next placeholder batch beyond the new epoch cmd
- return epoch_info snapshot

Signed-off-by: Oren Leiman <oren.leiman@redpanda.com>

Author: oleiman

src/v/cloud_topics/frontend/BUILD

@@ -40,6 +40,7 @@ redpanda_cc_library(
         "//src/v/cloud_topics:data_plane_api",
         "//src/v/cloud_topics:log_reader_config",
         "//src/v/cloud_topics:state_accessors",
+        "//src/v/cloud_topics:types",
         "//src/v/cloud_topics/level_one/frontend_reader:reader",
         "//src/v/cloud_topics/level_zero/common:extent_meta",
         "//src/v/cloud_topics/level_zero/frontend_reader",

src/v/cloud_topics/frontend/frontend.cc

@@ -18,6 +18,7 @@
 #include "cloud_topics/level_zero/common/producer_queue.h"
 #include "cloud_topics/level_zero/frontend_reader/level_zero_reader.h"
 #include "cloud_topics/level_zero/stm/ctp_stm.h"
+#include "cloud_topics/level_zero/stm/ctp_stm_api.h"
 #include "cloud_topics/level_zero/stm/placeholder.h"
 #include "cloud_topics/logger.h"
 #include "cloud_topics/state_accessors.h"
@@ -1044,6 +1045,41 @@ frontend::epoch_info frontend::get_epoch_info() const {
     };
 }
 
+auto frontend::advance_epoch(
+  cloud_topics::cluster_epoch new_epoch,
+  model::timeout_clock::time_point deadline)
+  -> ss::future<std::expected<epoch_info, frontend_errc>> {
+    vlog(cd_log.debug, "{}: advance epoch to {}", ntp(), new_epoch);
+
+    constexpr auto api_errc_to_fe_errc =
+      [](ctp_stm_api_errc ec) -> frontend_errc {
+        switch (ec) {
+            using enum ctp_stm_api_errc;
+        case not_leader:
+            return frontend_errc::not_leader_for_partition;
+        case shutdown:
+        case failure:
+        case timeout:
+            return frontend_errc::timeout;
+        }
+    };
+
+    ss::abort_source as;
+
+    auto result = co_await _ctp_stm_api->advance_epoch(new_epoch, deadline, as);
+    if (!result.has_value()) {
+        co_return std::unexpected{api_errc_to_fe_errc(result.error())};
+    }
+
+    auto adv_res = co_await _ctp_stm_api->sync_to_next_placeholder(
+      deadline, as);
+    if (!adv_res.has_value()) {
+        co_return std::unexpected{api_errc_to_fe_errc(adv_res.error())};
+    }
+
+    co_return get_epoch_info();
+}
+
 fmt::iterator
 frontend::coarse_grained_timequery_result::format_to(fmt::iterator it) const {
     return fmt::format_to(

src/v/cloud_topics/frontend/frontend.h

@@ -167,6 +167,11 @@ class frontend final {
     /// Return current epoch state.
     epoch_info get_epoch_info() const;
 
+    /// Advance the partition to the current cluster epoch and return epoch
+    /// state.
+    ss::future<std::expected<epoch_info, frontend_errc>> advance_epoch(
+      cloud_topics::cluster_epoch, model::timeout_clock::time_point);
+
 private:
     // All timequeries work by first getting a coarse grained timequery result
     // from metadata indexes, then getting an exact answer using the datapath.

src/v/cloud_topics/frontend/tests/frontend_test.cc

@@ -188,3 +188,63 @@ TEST_F(frontend_fixture, test_replicate_epoch) {
         ASSERT_FALSE(res.has_value());
     }
 }
+
+TEST_F(frontend_fixture, test_advance_epoch) {
+    // This test verifies that frontend::advance_epoch() correctly integrates
+    // with the underlying ctp_stm_api to advance the partition's epoch and
+    // return consistent epoch_info.
+    const model::topic topic_name("advance_epoch_test");
+    model::ntp ntp(model::kafka_namespace, topic_name, 0);
+
+    cluster::topic_properties props;
+    props.cloud_topic_enabled = true;
+    props.shadow_indexing = model::shadow_indexing_mode::disabled;
+
+    add_topic({model::kafka_namespace, topic_name}, 1, props).get();
+    wait_for_leader(ntp).get();
+
+    auto partition = app.partition_manager.local().get(ntp);
+    ASSERT_TRUE(
+      partition->raft()->stm_manager()->get<cloud_topics::ctp_stm>()
+      != nullptr);
+
+    cloud_topics::frontend frontend(std::move(partition), _data_plane.get());
+
+    // Initially, get_epoch_info should return min epochs (no data yet)
+    auto initial_info = frontend.get_epoch_info();
+    EXPECT_EQ(initial_info.max_applied_epoch, cluster_epoch::min());
+    EXPECT_EQ(initial_info.estimated_inactive_epoch, cluster_epoch::min());
+
+    // Call advance_epoch to establish epoch 5
+    auto first_advance
+      = frontend.advance_epoch(cluster_epoch(5), model::no_timeout).get();
+    ASSERT_TRUE(first_advance.has_value())
+      << "first advance_epoch should succeed";
+    EXPECT_EQ(first_advance.value().max_applied_epoch, cluster_epoch(5));
+    // first advance call reflects the new epoch right away because the stm
+    // state was empty
+    EXPECT_EQ(first_advance.value().estimated_inactive_epoch, cluster_epoch(4));
+
+    // Call advance_epoch with a higher epoch (10)
+    auto advance_result
+      = frontend.advance_epoch(cluster_epoch(10), model::no_timeout).get();
+    ASSERT_TRUE(advance_result.has_value())
+      << "advance_epoch should succeed on leader";
+
+    auto epoch_info = advance_result.value();
+    // max_applied_epoch should now be 10, inactive epoch is 4 because
+    // prev_applied_epoch is 5
+    EXPECT_EQ(epoch_info.max_applied_epoch, cluster_epoch(10));
+    EXPECT_EQ(epoch_info.estimated_inactive_epoch, cluster_epoch(4));
+
+    // advance the epoch one more time to observe lower bound sync behavior
+
+    auto final_result
+      = frontend.advance_epoch(cluster_epoch(15), model::no_timeout).get();
+    ASSERT_TRUE(advance_result.has_value())
+      << "advance_epoch should succeed on leader";
+    auto final_info = final_result.value();
+    EXPECT_EQ(final_info.max_applied_epoch, cluster_epoch(15));
+    EXPECT_EQ(final_info.estimated_inactive_epoch, cluster_epoch(9));
+    EXPECT_EQ(final_info, frontend.get_epoch_info());
+}