RFC ts-ct-migration: archival flag + structural routing [PR 4]

src/v/cloud_storage/tests/cloud_storage_e2e_test.cc

@@ -18,6 +18,7 @@
 #include "cluster/controller_api.h"
 #include "cluster/health_monitor_frontend.h"
 #include "kafka/client/transport.h"
+#include "kafka/data/partition_proxy.h"
 #include "kafka/data/replicated_partition.h"
 #include "kafka/protocol/fetch.h"
 #include "kafka/protocol/find_coordinator.h"
@@ -198,6 +199,166 @@ TEST_F(ManualFixture, TestSizeEstimationWithCloud) {
     }
 }
 
+// A partition mid tiered->cloud migration is served transparently as tiered
+// storage. The migration is entered by the operator flipping the topic's
+// storage mode to cloud (topic_mode == cloud), but the durable partition_mode
+// lags at tiered until cutover, and the serving accessors key on partition_mode
+// -- so cloud_topic_enabled()/is_remote_fetch_enabled() still report the tiered
+// values and nothing about the read path changes. This exercises the full
+// replicated_partition API surface plus make_partition_proxy routing and
+// asserts byte-for-byte parity with the same partition before the flip:
+// offsets, size estimates, and timequery are unchanged, prefix truncation still
+// works, and the partition still routes to the replicated_partition path.
+// Regression for the timequery corner -- it must consult the tiered/imported
+// prefix, not the local-log start, throughout.
+TEST_F(ManualFixture, MigratingPartitionServedAsTieredStorage) {
+    test_local_cfg.get("cloud_storage_disable_upload_loop_for_tests")
+      .set_value(true);
+    test_local_cfg.get("cloud_storage_spillover_manifest_max_segments")
+      .set_value(std::make_optional<size_t>(5));
+    test_local_cfg.get("cloud_storage_spillover_manifest_size")
+      .set_value(std::optional<size_t>{});
+    const model::topic topic_name("migrating-rp");
+    model::ntp ntp(model::kafka_namespace, topic_name, 0);
+
+    cluster::topic_properties props;
+    props.shadow_indexing = model::shadow_indexing_mode::full;
+    props.cleanup_policy_bitflags = model::cleanup_policy_bitflags::deletion;
+    add_topic({model::kafka_namespace, topic_name}, 1, props).get();
+    wait_for_leader(ntp).get();
+
+    auto partition = app.partition_manager.local().get(ntp);
+    auto& archiver = partition->archiver().value().get();
+    archiver.initialize_probe();
+    tests::remote_segment_generator gen(make_kafka_client().get(), *partition);
+    auto deferred_g_close = ss::defer([&gen] { gen.stop().get(); });
+    auto total_records = gen.num_segments(40)
+                           .batches_per_segment(5)
+                           .additional_local_segments(10)
+                           .produce()
+                           .get();
+    ASSERT_GE(total_records, 200);
+    ASSERT_TRUE(archiver.sync_for_tests().get());
+    archiver.apply_spillover().get();
+
+    // Keep some segments local and the prefix only in tiered storage, so the
+    // API spans both regions (cloud prefix + local suffix) and the parity check
+    // is meaningful.
+    auto log = partition->log();
+    auto& manifest = partition->archival_meta_stm()->manifest();
+    log->set_cloud_gc_offset(model::next_offset(manifest.get_last_offset()));
+    // Wait for GC to trim the local prefix so it lives only in tiered storage
+    // (the local log start advances above 0). Reads of the prefix must then
+    // come from the cloud, which is what makes the migrating-parity check below
+    // meaningful (and is the condition the timequery bug needed to manifest).
+    RPTEST_REQUIRE_EVENTUALLY(
+      20s, [&] { return log->offsets().start_offset > model::offset(0); });
+
+    struct snapshot {
+        model::offset start;
+        model::offset hwm;
+        model::offset local_start;
+        model::offset lso;
+        kafka::leader_epoch epoch;
+        model::offset offset_lag;
+        size_t total_sz;
+        size_t cloud_sz;
+        std::optional<model::offset> tq_earliest;
+    };
+    auto capture = [](kafka::replicated_partition& rp) {
+        auto lso = rp.last_stable_offset();
+        EXPECT_FALSE(lso.has_error());
+        auto last = kafka::prev_offset(model::offset_cast(lso.value()));
+        auto local_start = model::offset_cast(rp.local_start_offset());
+        // timequery for the epoch resolves to the earliest available offset; it
+        // must consult the tiered/imported prefix, not just the local log.
+        storage::timequery_config tq_cfg{
+          rp.start_offset(),
+          model::timestamp(0),
+          rp.high_watermark(),
+          {model::record_batch_type::raft_data},
+          std::nullopt};
+        auto tq = rp.timequery(tq_cfg).get();
+        return snapshot{
+          .start = rp.start_offset(),
+          .hwm = rp.high_watermark(),
+          .local_start = rp.local_start_offset(),
+          .lso = lso.value(),
+          .epoch = rp.leader_epoch(),
+          .offset_lag = rp.offset_lag(),
+          .total_sz = rp.estimate_size_between(kafka::offset(0), last),
+          .cloud_sz = rp.estimate_size_between(
+            kafka::offset(0), kafka::prev_offset(local_start)),
+          .tq_earliest = tq.has_value()
+                           ? std::optional<model::offset>(tq->offset)
+                           : std::nullopt,
+        };
+    };
+
+    kafka::replicated_partition rp_tiered(partition);
+    auto tiered = capture(rp_tiered);
+    // The prefix really is cloud-only (local log starts above 0), so timequery
+    // for the epoch must reach into tiered storage to answer 0.
+    ASSERT_GT(tiered.local_start, model::offset(0));
+    ASSERT_EQ(
+      tiered.tq_earliest, std::optional<model::offset>(model::offset(0)));
+
+    // Enter the migrating state: the operator flips the topic's storage mode to
+    // cloud (topic_mode), while the durable partition_mode lags at tiered until
+    // cutover. Serving accessors key on partition_mode, so
+    // cloud_topic_enabled() stays false and the partition keeps being served as
+    // tiered storage.
+    ASSERT_TRUE(log->config().has_overrides());
+    auto overrides = log->config().get_overrides();
+    overrides.storage_mode = model::redpanda_storage_mode::cloud;
+    log->set_overrides(overrides);
+    log->set_partition_mode(model::redpanda_storage_mode::tiered);
+    ASSERT_EQ(log->config().topic_mode(), model::redpanda_storage_mode::cloud);
+    ASSERT_EQ(
+      log->config().partition_mode(), model::redpanda_storage_mode::tiered);
+    ASSERT_FALSE(log->config().cloud_topic_enabled());
+
+    kafka::replicated_partition rp_mig(partition);
+    auto migrating = capture(rp_mig);
+
+    // Transparent migration: every read/size API returns exactly what it did as
+    // plain tiered storage.
+    EXPECT_EQ(tiered.start, migrating.start);
+    EXPECT_EQ(tiered.hwm, migrating.hwm);
+    EXPECT_EQ(tiered.local_start, migrating.local_start);
+    EXPECT_EQ(tiered.lso, migrating.lso);
+    EXPECT_EQ(tiered.epoch, migrating.epoch);
+    EXPECT_EQ(tiered.offset_lag, migrating.offset_lag);
+    // Entering the migrating state writes nothing to the log (it only adjusts
+    // the ntp_config overrides), so every size estimate is exactly unchanged.
+    EXPECT_EQ(tiered.total_sz, migrating.total_sz);
+    EXPECT_EQ(tiered.cloud_sz, migrating.cloud_sz);
+    // Regression: timequery still resolves to offset 0 via the imported extents
+    // (not the local-log start) while migrating.
+    EXPECT_EQ(tiered.tq_earliest, migrating.tq_earliest);
+    EXPECT_EQ(
+      migrating.tq_earliest, std::optional<model::offset>(model::offset(0)));
+
+    // make_partition_proxy routes the migrating partition to the tiered-storage
+    // (replicated_partition) path via the standard cloud_topic_enabled()==false
+    // branch -- partition_mode lags at tiered, so no manifest-specific routing
+    // gate is needed. It must behave identically to a directly-constructed
+    // replicated_partition, not the (empty) cloud-topic path.
+    kafka::partition_proxy proxy = kafka::make_partition_proxy(partition);
+    EXPECT_EQ(proxy.start_offset(), rp_mig.start_offset());
+    EXPECT_EQ(proxy.high_watermark(), rp_mig.high_watermark());
+    EXPECT_GT(proxy.high_watermark(), model::offset(0));
+
+    // DeleteRecords / prefix truncation works while migrating and advances the
+    // start offset (eviction STM + archival manifest head-prune path).
+    auto trunc = model::offset(total_records / 2);
+    auto tec
+      = rp_mig.prefix_truncate(trunc, ss::lowres_clock::now() + 30s).get();
+    ASSERT_EQ(tec, kafka::error_code::none);
+    RPTEST_REQUIRE_EVENTUALLY(
+      10s, [&] { return rp_mig.start_offset() >= trunc; });
+}
+
 class EndToEndFixture
   : public s3_imposter_fixture
   , public manual_metadata_upload_mixin

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/BUILD

@@ -1407,6 +1407,7 @@ redpanda_cc_library(
         ":types",
         "//src/v/base",
         "//src/v/storage",
+        "//src/v/utils:prefix_logger",
         "@fmt",
         "@seastar",
     ],