archival: clamp local-log trim while the manifest holds archived data

A cloud-mode partition whose storage mode has been flipped reports
is_archival_enabled() == false, which would make max_removable_local_log_offset
collect_all and stop constraining local-log truncation -- evicting
tiered-storage data not yet uploaded. Keep constraining via
cloud_recoverable_offset() while the partition still holds archived data (a
non-empty live manifest or a spillover archive), introduced here as
holds_archived_data(); once the manifest is emptied, trimming resumes.

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

Author: sjust-redpanda

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();
 }

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;