Merge pull request #29921 from vbotbuildovich/backport-pr-29831-v25.3.x-535

[v25.3.x] archival: repair misaligned archive offsets

Author: nvartolomei

src/v/cloud_storage/partition_manifest.cc

@@ -1152,6 +1152,11 @@ void partition_manifest::spillover(const segment_meta& spillover_meta) {
 }
 
 segment_meta partition_manifest::make_manifest_metadata() const {
+    if (empty()) {
+        throw std::runtime_error(
+          "can't make manifest metadata for empty manifest");
+    }
+
     return segment_meta{
       .size_bytes = cloud_log_size(),
       .base_offset = get_start_offset().value(),
@@ -2811,6 +2816,69 @@ void partition_manifest::process_anomalies(
       _last_scrubbed_offset);
 }
 
+std::optional<partition_manifest> partition_manifest::repair_state() const {
+    auto repaired = do_repair_state();
+
+    // Guard against "infinite repair loop" scenario, where the manifest is
+    // repaired but the repaired manifest is still inconsistent and needs to be
+    // repaired again.
+    if (repaired.has_value() && repaired->do_repair_state().has_value()) {
+        throw std::runtime_error(
+          fmt::format(
+            "[{}] Manifest repair failed: repaired manifest is still "
+            "inconsistent, this should never happen",
+            display_name()));
+    }
+
+    return repaired;
+}
+
+std::optional<partition_manifest> partition_manifest::do_repair_state() const {
+    if (!get_spillover_map().empty()) {
+        const auto first_spill = get_spillover_map().begin();
+        if (
+          _archive_start_offset < first_spill->base_offset
+          || _archive_clean_offset < first_spill->base_offset) {
+            auto cloned = clone();
+
+            // Old versions of redpanda (pre v25.3) could have a manifest with
+            // incorrect `archive_start_offset` and `archive_clean_offset`
+            // values—below the offset of the first spillover manifest. To avoid
+            // issues with such manifests, we need to detect this case and fix
+            // the offsets by bumping them up to the offset of the first
+            // spillover manifest.
+
+            if (_archive_start_offset < first_spill->base_offset) {
+                vlog(
+                  cst_log.warn,
+                  "[{}] archive_start_offset {} is below the first spillover "
+                  "manifest offset {}, advancing it to align with the first "
+                  "spillover manifest offset.",
+                  display_name(),
+                  _archive_start_offset,
+                  first_spill->base_offset);
+                cloned.set_archive_start_offset(
+                  first_spill->base_offset, first_spill->delta_offset);
+            }
+            if (_archive_clean_offset < first_spill->base_offset) {
+                vlog(
+                  cst_log.warn,
+                  "[{}] archive_clean_offset {} is below the first spillover "
+                  "manifest offset {}, advancing it to align with the first "
+                  "spillover manifest offset.",
+                  display_name(),
+                  _archive_clean_offset,
+                  first_spill->base_offset);
+                cloned.set_archive_clean_offset(first_spill->base_offset, 0);
+            }
+
+            return cloned;
+        }
+    }
+
+    return std::nullopt;
+}
+
 std::ostream& operator<<(std::ostream& o, const partition_manifest& pm) {
     o << "{manifest: ";
     pm.serialize_json(o, false);

src/v/cloud_storage/partition_manifest.h

@@ -381,6 +381,8 @@ class partition_manifest : public base_manifest {
     /// This mechanism is used by the 'spillover' mechanism. The 'segment_meta'
     /// instances that describe spillover manifests are stored using this
     /// format.
+    ///
+    /// \pre The manifest must not be empty.
     segment_meta make_manifest_metadata() const;
 
     /// Return 'true' if the spillover manifest can be added to
@@ -594,6 +596,10 @@ class partition_manifest : public base_manifest {
       scrub_status status,
       anomalies detected);
 
+    /// Returns a repaired copy of the manifest if the manifest has known
+    /// defects. Returns `std::nullopt` otherwise.
+    std::optional<partition_manifest> repair_state() const;
+
 private:
     ss::sstring display_name() const;
     std::optional<kafka::offset> compute_start_kafka_offset_local() const;
@@ -643,6 +649,8 @@ class partition_manifest : public base_manifest {
     void serialize_removed_segment_meta(
       const lw_segment_meta& meta, serialization_cursor_ptr cursor) const;
 
+    std::optional<partition_manifest> do_repair_state() const;
+
     model::ntp _ntp;
     model::initial_revision_id _rev;
 

src/v/cloud_storage/tests/partition_manifest_test.cc

@@ -2844,3 +2844,96 @@ SEASTAR_THREAD_TEST_CASE(
     BOOST_REQUIRE(restored == m);
     BOOST_REQUIRE(m.get_applied_offset() == model::offset{100});
 }
+
+SEASTAR_THREAD_TEST_CASE(test_repair_state_no_spillovers) {
+    partition_manifest m;
+    m.add(make_segment({10, 99}, 1234));
+    m.add(make_segment({100, 199}, 1000));
+
+    BOOST_REQUIRE(!m.repair_state().has_value());
+}
+
+SEASTAR_THREAD_TEST_CASE(test_repair_state_aligned) {
+    partition_manifest m;
+    m.add(make_segment({10, 199}, 2234, 0));
+    m.add(make_segment({200, 299}, 2000, 0));
+    m.add(make_segment({300, 399}, 500, 0));
+    {
+        auto spill_manifest = partition_manifest{
+          m.get_ntp(), m.get_revision_id()};
+        spill_manifest.add(*m.begin());
+        m.spillover(spill_manifest.make_manifest_metadata());
+        m.set_archive_start_offset(model::offset{10}, model::offset_delta{0});
+        m.set_archive_clean_offset(model::offset{10}, 0);
+    }
+
+    auto first_spill_base = m.get_spillover_map().begin()->base_offset;
+    BOOST_REQUIRE_EQUAL(first_spill_base, model::offset{10});
+    BOOST_REQUIRE(!m.repair_state().has_value());
+}
+
+SEASTAR_THREAD_TEST_CASE(test_repair_state_misaligned) {
+    partition_manifest m;
+    m.add(make_segment({10, 199}, 2234, 0));
+    m.add(make_segment({200, 299}, 2000, 0));
+    m.add(make_segment({300, 399}, 500, 0));
+
+    {
+        auto spill_manifest = partition_manifest{
+          m.get_ntp(), m.get_revision_id()};
+        spill_manifest.add(*m.begin());
+        m.spillover(spill_manifest.make_manifest_metadata());
+        m.set_archive_start_offset(model::offset{5}, model::offset_delta{0});
+        m.set_archive_clean_offset(model::offset{5}, 0);
+    }
+
+    auto first_spill = *m.get_spillover_map().begin();
+    BOOST_REQUIRE(m.get_archive_start_offset() < first_spill.base_offset);
+
+    auto repaired = m.repair_state();
+    BOOST_REQUIRE(repaired.has_value());
+    BOOST_REQUIRE_EQUAL(
+      repaired->get_archive_start_offset(), first_spill.base_offset);
+    BOOST_REQUIRE_EQUAL(
+      repaired->get_archive_clean_offset(), first_spill.base_offset);
+
+    // Original manifest is unchanged.
+    BOOST_REQUIRE_EQUAL(m.get_archive_start_offset(), model::offset{5});
+    BOOST_REQUIRE_EQUAL(m.get_archive_clean_offset(), model::offset{5});
+
+    // Repaired manifest no longer needs repair.
+    BOOST_REQUIRE(!repaired->repair_state().has_value());
+}
+
+SEASTAR_THREAD_TEST_CASE(test_repair_state_misaligned_clean_offset) {
+    partition_manifest m;
+    m.add(make_segment({10, 199}, 2234, 0));
+    m.add(make_segment({200, 299}, 2000, 0));
+    m.add(make_segment({300, 399}, 500, 0));
+
+    {
+        auto spill_manifest = partition_manifest{
+          m.get_ntp(), m.get_revision_id()};
+        spill_manifest.add(*m.begin());
+        m.spillover(spill_manifest.make_manifest_metadata());
+        m.set_archive_start_offset(model::offset{200}, model::offset_delta{0});
+        m.set_archive_clean_offset(model::offset{5}, 0);
+    }
+
+    auto first_spill = *m.get_spillover_map().begin();
+    BOOST_REQUIRE(m.get_archive_clean_offset() < first_spill.base_offset);
+
+    auto repaired = m.repair_state();
+    BOOST_REQUIRE(repaired.has_value());
+    BOOST_REQUIRE_EQUAL(
+      repaired->get_archive_start_offset(), model::offset{200});
+    BOOST_REQUIRE_EQUAL(
+      repaired->get_archive_clean_offset(), first_spill.base_offset);
+
+    // Original manifest is unchanged.
+    BOOST_REQUIRE_EQUAL(m.get_archive_start_offset(), model::offset{200});
+    BOOST_REQUIRE_EQUAL(m.get_archive_clean_offset(), model::offset{5});
+
+    // Repaired manifest no longer needs repair.
+    BOOST_REQUIRE(!repaired->repair_state().has_value());
+}

src/v/cluster/archival/ntp_archiver_service.cc

@@ -420,6 +420,40 @@ archival_stm_fence ntp_archiver::emit_rw_fence() {
     };
 }
 
+ss::future<std::error_code>
+ntp_archiver::maybe_repair_manifest(ss::lowres_clock::time_point deadline) {
+    auto repaired_copy = _parent.archival_meta_stm()->manifest().repair_state();
+    if (!repaired_copy) {
+        co_return std::error_code{};
+    }
+
+    vlog(
+      _rtclog.warn, "Manifest repair created a new manifest. Replicating it.");
+    auto batch = _parent.archival_meta_stm()->batch_start(deadline, _as);
+    auto fence = emit_rw_fence();
+    if (fence.emit_rw_fence_cmd) {
+        vlog(
+          _rtclog.debug,
+          "replace manifest with repair, read-write fence: {}",
+          fence.read_write_fence);
+        batch.read_write_fence(fence.read_write_fence);
+    }
+
+    batch.replace_manifest(repaired_copy->to_iobuf());
+    auto ec = co_await batch.replicate();
+    if (ec) {
+        vlog(
+          _rtclog.error,
+          "Failed to replace manifest with repaired version: {}",
+          ec);
+    } else {
+        vlog(
+          _rtclog.debug, "Finished replacing manifest with repaired version");
+    }
+
+    co_return ec;
+}
+
 void ntp_archiver::log_collected_traces() noexcept {
     try {
         _rtclog.bypass_tracing([this] {
@@ -700,6 +734,13 @@ ss::future<> ntp_archiver::upload_until_abort() {
             }
         }
 
+        if (auto ec = co_await maybe_repair_manifest(
+              ss::lowres_clock::now() + sync_timeout);
+            ec) {
+            vlog(_rtclog.warn, "Failed to repair manifest: {}, retrying", ec);
+            continue;
+        }
+
         vlog(_rtclog.debug, "upload loop synced in term {}", _start_term);
         if (!may_begin_uploads()) {
             continue;

src/v/cluster/archival/ntp_archiver_service.h

@@ -482,6 +482,9 @@ class ntp_archiver {
     /// Create a fence value for the next STM operation
     archival_stm_fence emit_rw_fence();
 
+    ss::future<std::error_code>
+    maybe_repair_manifest(ss::lowres_clock::time_point deadline);
+
     /// Delete objects, return true on success and false otherwise
     ss::future<bool>
     batch_delete(std::vector<cloud_storage_clients::object_key> paths);