tests: add replicated_partition mid-migration API-parity unit test

Exercises the replicated_partition API surface on a cloud-storage-backed
partition mid tiered->cloud migration (cloud_topic_enabled + non-empty archival
manifest + migration flag) and asserts it matches the same partition as plain
tiered storage: offsets (start / high_watermark / last_stable / local_start /
leader_epoch / offset_lag), size estimates, and timequery are unchanged when the
storage mode flips to cloud, and prefix truncation works and advances the start.

Complements the e2e API battery with per-method C++ coverage -- including the
non-Kafka-reachable methods (estimate_size_between) -- and regresses the
timequery fix (timequery must consult the imported extents, not just the local
log, while migrating).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: sjust-redpanda

src/v/cloud_storage/tests/cloud_storage_e2e_test.cc

@@ -198,6 +198,154 @@ TEST_F(ManualFixture, TestSizeEstimationWithCloud) {
     }
 }
 
+// Exercises the replicated_partition API surface on a partition mid
+// tiered->cloud migration (cloud_topic_enabled + non-empty archival manifest +
+// migration flag) and asserts it behaves identically to the same partition as
+// plain tiered storage. Migration is transparent for a TS-served partition, so
+// offsets, size estimates, and timequery must not change when the storage mode
+// flips to cloud. Regression for the make_partition_proxy routing /
+// storage_mode-vs-manifest corner -- e.g. timequery once consulted only the
+// local log mid-migration and returned the local-log start instead of the true
+// earliest offset.
+TEST_F(ManualFixture, ReplicatedPartitionMidMigrationApiParity) {
+    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)));
+
+    // Flip the partition into the migrating state: cloud storage mode
+    // (cloud_topic_enabled) + the archival STM migration flag, manifest still
+    // non-empty.
+    ASSERT_TRUE(log->config().has_overrides());
+    auto overrides = log->config().get_overrides();
+    overrides.storage_mode = model::redpanda_storage_mode::cloud;
+    log->set_overrides(overrides);
+    ss::abort_source as;
+    auto mec = partition->archival_meta_stm()
+                 ->set_migration_state(true, ss::lowres_clock::now() + 30s, as)
+                 .get();
+    ASSERT_FALSE(mec) << "set_migration_state failed: " << mec.message();
+    ASSERT_TRUE(log->config().cloud_topic_enabled());
+    ASSERT_TRUE(partition->archival_meta_stm()->is_migrating());
+
+    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);
+    // total_sz spans the local region, where flipping to migrating appended a
+    // set_migration_state control batch to the active segment; allow that small
+    // delta. cloud_sz reads only the (unchanged) manifest, so stays exact.
+    EXPECT_NEAR(tiered.total_sz, migrating.total_sz, 4096);
+    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)));
+
+    // 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