RFC: ts-ct-migration: STM coexistence [PR 2]

src/v/cloud_topics/level_zero/stm/BUILD

@@ -182,6 +182,7 @@ redpanda_cc_library(
     implementation_deps = [
         ":ctp_stm",
         "//src/v/cloud_topics:logger",
+        "//src/v/config",
     ],
     visibility = ["//visibility:public"],
     deps = ["//src/v/cluster:state_machine_registry"],

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

@@ -424,6 +424,16 @@ ctp_stm::take_local_snapshot(ssx::semaphore_units) {
 }
 
 ss::future<> ctp_stm::apply_raft_snapshot(const iobuf& buf) {
+    // An empty snapshot is applied during recovery fast-forward: the raft
+    // snapshot created when bootstrapping a pre-existing partition carries no
+    // per-STM data, so state_machine_manager hands each STM an empty buffer to
+    // advance over. There is no ctp_stm state to restore then -- e.g. a
+    // partition recovered mid tiered->cloud migration has no L1 state (its data
+    // is still in tiered storage). Keep the default (empty) state and advance,
+    // mirroring log_eviction_stm's empty-snapshot handling.
+    if (buf.empty()) {
+        co_return;
+    }
     auto snap = serde::from_iobuf<ctp_stm_snapshot>(buf.copy());
     _state = std::move(snap.state);
     _epoch_checker = snap.checker;

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

@@ -121,13 +121,21 @@ ss::future<std::expected<std::monostate, ctp_stm_api_errc>>
 ctp_stm_api::advance_reconciled_offset(
   kafka::offset lro,
   model::timeout_clock::time_point deadline,
+  ss::abort_source& as) {
+    auto lrlo = _stm->_raft->log()->to_log_offset(kafka::offset_cast(lro));
+    return advance_reconciled_offset(lro, lrlo, deadline, as);
+}
+
+ss::future<std::expected<std::monostate, ctp_stm_api_errc>>
+ctp_stm_api::advance_reconciled_offset(
+  kafka::offset lro,
+  model::offset lrlo,
+  model::timeout_clock::time_point deadline,
   ss::abort_source& as) {
     if (lro <= get_last_reconciled_offset()) {
         co_return std::monostate{};
     }
 
-    auto lrlo = _stm->_raft->log()->to_log_offset(kafka::offset_cast(lro));
-
     vlog(
       _log.debug,
       "Replicating ctp_stm_cmd::advance_reconciled_offset{{lro:{} lrlo:{}}}",

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

@@ -76,6 +76,19 @@ class ctp_stm_api {
       model::timeout_clock::time_point deadline,
       ss::abort_source& as);
 
+    /// Advance the reconciled offset with an explicit log offset, rather than
+    /// deriving it from the raft offset translator. Used to seed the
+    /// reconciliation baseline of a TS-migrated partition directly to the
+    /// migration boundary (kafka offset + the raft offset of the last uploaded
+    /// TS segment) so the already-uploaded TS region of the local raft log can
+    /// be trimmed without waiting for the first CT reconciliation cycle.
+    ss::future<std::expected<std::monostate, ctp_stm_api_errc>>
+    advance_reconciled_offset(
+      kafka::offset last_reconciled_offset,
+      model::offset last_reconciled_log_offset,
+      model::timeout_clock::time_point deadline,
+      ss::abort_source& as);
+
     ss::future<std::expected<std::monostate, ctp_stm_api_errc>>
     set_start_offset(
       kafka::offset new_start_offset,

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

@@ -12,13 +12,25 @@
 
 #include "cloud_topics/level_zero/stm/ctp_stm.h"
 #include "cloud_topics/logger.h"
+#include "config/configuration.h"
 
 namespace cloud_topics::l0 {
 
 bool ctp_stm_factory::is_applicable_for(
   const storage::ntp_config& ntp_cfg) const {
-    return ntp_cfg.cloud_topic_enabled()
-           && !ntp_cfg.is_read_replica_mode_enabled();
+    if (ntp_cfg.is_read_replica_mode_enabled()) {
+        return false;
+    }
+    if (ntp_cfg.cloud_topic_enabled()) {
+        return true;
+    }
+    // Pre-install an (idle) ctp_stm on tiered-storage partitions when cloud
+    // topics are available, so a tiered->cloud/tiered_cloud migration needs no
+    // runtime STM install -- the STM is already present and inert
+    // (get_max_collectible_offset() returns max() until CT data is applied)
+    // until the partition becomes a cloud topic.
+    return config::shard_local_cfg().cloud_storage_enabled()
+           && ntp_cfg.is_archival_enabled();
 }
 
 void ctp_stm_factory::create(

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

@@ -151,7 +151,16 @@ model::offset ctp_stm_state::get_max_collectible_offset() const noexcept {
     if (_last_reconciled_log_offset.has_value()) {
         return _last_reconciled_log_offset.value();
     }
-    // Truncation is impossible without LRO
+    // An STM that has applied no CT data must not constrain truncation: it may
+    // be an inert passenger on a partition that is not (yet) a cloud topic
+    // (e.g. a tiered partition pre-installed with ctp_stm so a migration needs
+    // no runtime STM install). Without this a passenger would pin the local log
+    // at offset::min() forever.
+    if (!_max_applied_epoch.has_value()) {
+        return model::offset::max();
+    }
+    // CT data has been applied but not yet reconciled to L1: protect it.
+    // Truncation is impossible without an LRO.
     return model::offset::min();
 }
 

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

@@ -30,7 +30,9 @@ TEST(ctp_stm_state_test, initial_state) {
     EXPECT_FALSE(state.get_max_seen_epoch(model::term_id(1)).has_value());
     EXPECT_FALSE(state.get_last_reconciled_offset().has_value());
     EXPECT_FALSE(state.get_last_reconciled_log_offset().has_value());
-    EXPECT_EQ(state.get_max_collectible_offset(), model::offset::min());
+    // An STM with no applied CT data is an inert passenger and must not
+    // constrain truncation.
+    EXPECT_EQ(state.get_max_collectible_offset(), model::offset::max());
 }
 
 TEST(ctp_stm_state_test, advance_max_seen_epoch) {
@@ -107,8 +109,15 @@ TEST(ctp_stm_state_test, advance_last_reconciled_offset) {
 TEST(ctp_stm_state_test, get_max_collectible_offset) {
     ct::ctp_stm_state state;
 
+    // No CT data applied yet: inert passenger, do not constrain truncation.
+    EXPECT_EQ(state.get_max_collectible_offset(), model::offset::max());
+
+    // CT data applied (a placeholder advanced the epoch) but not yet reconciled
+    // to L1: protect it by pinning collection at min().
+    state.advance_epoch(ct::cluster_epoch(1), model::offset(10));
     EXPECT_EQ(state.get_max_collectible_offset(), model::offset::min());
 
+    // Once reconciled, collect up to the reconciled log offset.
     model::offset log_offset(500);
     state.advance_last_reconciled_offset(kafka::offset(300), log_offset);
     EXPECT_EQ(state.get_max_collectible_offset(), log_offset);

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

@@ -48,6 +48,10 @@ struct ctp_stm_accessor {
           snapshot.header, std::move(snapshot.data));
     }
 
+    auto apply_raft_snapshot(ctp_stm& stm, const iobuf& buf) {
+        return stm.apply_raft_snapshot(buf);
+    }
+
     bool epoch_cv_has_waiters(ctp_stm& stm) {
         return stm._epoch_updated_cv.has_waiters();
     }
@@ -521,6 +525,37 @@ TEST_F_CORO(ctp_stm_fixture, test_snapshot) {
     }
 }
 
+// The recovery fast-forward hands each STM an empty raft snapshot to advance
+// over: a bootstrapped pre-existing partition carries no per-STM data, and a
+// partition recovered mid tiered->cloud migration has no L1 ctp_stm state (its
+// data is still in tiered storage). apply_raft_snapshot must treat an empty
+// buffer as a no-op -- not try to deserialize it (which threw before the guard)
+// and not clobber existing state with a default.
+TEST_F_CORO(ctp_stm_fixture, apply_empty_raft_snapshot_is_noop) {
+    co_await start();
+    co_await wait_for_leader(raft::default_timeout());
+    auto& leader = node(*get_leader());
+
+    // Establish some applied state so we can prove the empty snapshot doesn't
+    // reset it.
+    auto b1 = make_record_batch(ct::cluster_epoch{2}, model::offset{0}, 0);
+    auto res = co_await replicate_record_batch(leader, std::move(b1));
+    ASSERT_TRUE_CORO(res.has_value());
+    auto max_epoch_before = api(leader).get_max_epoch();
+    ASSERT_TRUE_CORO(max_epoch_before.has_value());
+    ASSERT_EQ_CORO(max_epoch_before.value(), ct::cluster_epoch{2});
+
+    auto stm = get_stm<0>(leader);
+    ct::ctp_stm_accessor a;
+    // Must not throw on an empty buffer.
+    co_await a.apply_raft_snapshot(*stm, iobuf{});
+
+    // State preserved, not reset to default.
+    auto max_epoch_after = api(leader).get_max_epoch();
+    ASSERT_TRUE_CORO(max_epoch_after.has_value());
+    ASSERT_EQ_CORO(max_epoch_after.value(), max_epoch_before.value());
+}
+
 TEST_F_CORO(ctp_stm_fixture, test_fence_epoch_concurrent_new_epoch) {
     // This test verifies the optimization in fence_epoch() where multiple
     // concurrent requests for a new epoch only require one write lock.

src/v/cluster/archival/archival_metadata_stm.cc

@@ -1373,6 +1373,18 @@ model::offset archival_metadata_stm::max_removable_local_log_offset() {
         collect_all = false;
     }
 
+    // The manifest, not the storage mode, governs local-log truncation. While
+    // the partition still holds archived data (a non-empty live manifest or a
+    // spillover archive) it is still served from tiered storage and may hold
+    // local data not yet uploaded, so its local log must not be trimmed past
+    // cloud_recoverable_offset(). is_archival_enabled() can report false while
+    // archived data is still present -- e.g. a partition mid tiered->cloud
+    // migration whose effective storage mode is cloud -- which would otherwise
+    // collect_all and stop constraining truncation, evicting the not-yet-
+    // uploaded data. Gate collect_all on holds_archived_data() so it takes
+    // effect only once the manifest is cleared.
+    collect_all = collect_all && !holds_archived_data();
+
     if (collect_all || is_read_replica || (uploads_paused && gaps_allowed)) {
         // The archival is disabled but the state machine still exists so we
         // shouldn't stop eviction from happening.
@@ -1690,6 +1702,11 @@ model::offset archival_metadata_stm::get_archive_start_offset() const {
     return _manifest->get_archive_start_offset();
 }
 
+bool archival_metadata_stm::holds_archived_data() const {
+    return _manifest->size() > 0
+           || _manifest->get_archive_start_offset() != model::offset{};
+}
+
 model::offset archival_metadata_stm::get_archive_clean_offset() const {
     return _manifest->get_archive_clean_offset();
 }
@@ -1729,10 +1746,14 @@ archival_metadata_stm_factory::archival_metadata_stm_factory(
 
 bool archival_metadata_stm_factory::is_applicable_for(
   const storage::ntp_config& ntp_cfg) const {
+    // The archival STM is created on cloud-topic partitions too (no
+    // cloud_topic_enabled() == false guard). For a partition migrated from
+    // tiered storage it is reconstructed from its snapshot and its manifest
+    // stays available to gate local-log truncation. For a partition that was
+    // always a cloud topic the manifest is empty, so it is an inert passenger.
     return _cloud_storage_enabled && _cloud_storage_api.local_is_initialized()
            && ntp_cfg.ntp().tp.topic != model::kafka_consumer_offsets_topic
-           && ntp_cfg.ntp().ns == model::kafka_namespace
-           && ntp_cfg.cloud_topic_enabled() == false;
+           && ntp_cfg.ntp().ns == model::kafka_namespace;
 }
 
 void archival_metadata_stm_factory::create(

src/v/cluster/archival/archival_metadata_stm.h

@@ -252,6 +252,10 @@ class archival_metadata_stm final : public raft::persisted_stm<> {
     model::offset get_last_offset() const;
     model::offset get_archive_start_offset() const;
     model::offset get_archive_clean_offset() const;
+
+    /// True if the partition still has data in tiered storage -- either in the
+    /// live STM manifest or offloaded to the spillover archive.
+    bool holds_archived_data() const;
     kafka::offset get_start_kafka_offset() const;
 
     // Return list of all segments that has to be

src/v/cluster/archival/tests/archival_metadata_stm_test.cc

@@ -256,6 +256,42 @@ FIXTURE_TEST(
       archival_stm->manifest().begin()->committed_offset, model::offset(99));
 }
 
+// With archival disabled (the flipped, mid-migration storage mode) but a
+// non-empty manifest, max_removable_local_log_offset keeps constraining
+// local-log truncation rather than collecting all -- protecting tiered-storage
+// data that has not yet been uploaded.
+FIXTURE_TEST(test_migration_local_trim_clamp, archival_metadata_stm_fixture) {
+    wait_for_confirmed_leader();
+
+    // The fixture's partition has archival disabled. With an empty manifest,
+    // nothing constrains local-log truncation.
+    BOOST_REQUIRE_EQUAL(
+      archival_stm->max_removable_local_log_offset(), model::offset::max());
+
+    // Add a segment so the manifest is non-empty.
+    std::vector<cloud_storage::segment_meta> m;
+    m.push_back(
+      segment_meta{
+        .base_offset = model::offset(0),
+        .committed_offset = model::offset(99),
+        .archiver_term = model::term_id(1),
+        .segment_term = model::term_id(1)});
+    archival_stm
+      ->add_segments(
+        m,
+        std::nullopt,
+        model::producer_id{},
+        ss::lowres_clock::now() + 10s,
+        never_abort,
+        cluster::segment_validated::yes)
+      .get();
+    BOOST_REQUIRE_EQUAL(archival_stm->manifest().size(), 1);
+
+    // Now truncation is constrained (no longer offset::max()).
+    BOOST_REQUIRE_NE(
+      archival_stm->max_removable_local_log_offset(), model::offset::max());
+}
+
 FIXTURE_TEST(test_archival_stm_segment_replace, archival_metadata_stm_fixture) {
     wait_for_confirmed_leader();
     std::vector<cloud_storage::segment_meta> m1;

src/v/cluster/log_eviction_stm.cc

@@ -384,9 +384,15 @@ log_eviction_stm_factory::log_eviction_stm_factory(storage::kvstore& kvstore)
 
 bool log_eviction_stm_factory::is_applicable_for(
   const storage::ntp_config& cfg) const {
-    if (cfg.cloud_topic_enabled()) {
-        return false;
-    }
+    // (No cloud_topic_enabled() == false guard.) Install the eviction STM on
+    // cloud-topic partitions too. STM membership is fixed at construction and a
+    // tiered->cloud migration does not reconstruct the partition, so a
+    // partition migrating from tiered storage -- still served from its local
+    // log + archival manifest -- must already carry the eviction STM to keep
+    // supporting prefix truncation (DeleteRecords) and retention while
+    // migrating. On a native cloud topic it is an inert passenger: retention
+    // and DeleteRecords go through the L1 path, so no local eviction requests
+    // are generated.
     return !storage::deletion_exempt(cfg.ntp());
 }