archival: Disable cross-term compaction

Currently, the segment_collector can collect data from segments
generated in different terms. The component is used to generate
compacted reuploads. This commit fixes this by disabling the cross term
segment collection.

Signed-off-by: Evgeny Lazin <[email protected]>

Author: Lazin

src/v/cluster/archival/segment_reupload.cc

@@ -10,6 +10,7 @@
 
 #include "segment_reupload.h"
 
+#include "base/vlog.h"
 #include "cloud_storage/partition_manifest.h"
 #include "config/configuration.h"
 #include "logger.h"
@@ -202,6 +203,8 @@ void segment_collector::do_collect(segment_collector_mode mode) {
 
     align_end_offset_to_manifest(
       _target_end_inclusive.value_or(current_segment_end));
+
+    align_with_term_boundary();
 }
 
 model::offset segment_collector::find_replacement_boundary() const {
@@ -273,6 +276,42 @@ void segment_collector::align_end_offset_to_manifest(
     }
 }
 
+void segment_collector::align_with_term_boundary() {
+    auto opt_term = _log.get_term(_begin_inclusive);
+    if (!opt_term.has_value()) {
+        vlog(
+          archival_log.warn,
+          "Can't find term for offset {}, the log could be truncated",
+          _begin_inclusive);
+        throw std::runtime_error(fmt_with_ctx(
+          fmt::format, "Can't find term for offset {}", _begin_inclusive));
+    }
+    auto opt_max_offset = _log.get_term_last_offset(opt_term.value());
+    if (!opt_max_offset.has_value()) {
+        vlog(
+          archival_log.warn,
+          "Can't find last offset for term {}",
+          opt_term.value());
+        throw std::runtime_error(fmt_with_ctx(
+          fmt::format, "Can't find last offset for term {}", opt_term.value()));
+    }
+    vlog(
+      archival_log.debug,
+      "align_with_term_boundary, begin={}, end={}, term={}, "
+      "last-term-offset={}",
+      _begin_inclusive,
+      _end_inclusive,
+      opt_term.value(),
+      opt_max_offset.value());
+    // Invariant: 'max_offset' can't be in the middle of the segment
+    // because we're rolling a segment when new term starts.
+    auto max_offset = opt_max_offset.value();
+    // NOTE: if the segment lookup overshoots we can return max_offset
+    // because it's guaranteed to be a committed offset of some segment
+    // in the manifest.
+    _end_inclusive = std::min(max_offset, _end_inclusive);
+}
+
 segment_collector::lookup_result segment_collector::find_next_segment(
   model::offset start_offset, segment_collector_mode mode) {
     // 'start_offset' should always be above the start offset of the local log

src/v/cluster/archival/segment_reupload.h

@@ -113,6 +113,9 @@ class segment_collector {
     /// overlap.
     void align_end_offset_to_manifest(model::offset compacted_segment_end);
 
+    /// Make sure that reupload begins and ends in the same term.
+    void align_with_term_boundary();
+
     /// Finds the offset which the collection needs to progress upto in order to
     /// replace at least one manifest segment. The collection is valid if it
     /// reaches the replacement boundary.

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

@@ -367,7 +367,7 @@ FIXTURE_TEST(test_upload_compacted_segments, reupload_fixture) {
 FIXTURE_TEST(test_upload_compacted_segments_concat, reupload_fixture) {
     std::vector<segment_desc> segments = {
       {manifest_ntp, model::offset(0), model::term_id(1), 1000, 2},
-      {manifest_ntp, model::offset(1000), model::term_id(4), 10, 2},
+      {manifest_ntp, model::offset(1000), model::term_id(1), 10, 2},
     };
 
     initialize(segments);
@@ -385,7 +385,7 @@ FIXTURE_TEST(test_upload_compacted_segments_concat, reupload_fixture) {
 
     auto manifest = verify_manifest_request(*part);
     verify_segment_request("0-1-v1.log", manifest);
-    verify_segment_request("1000-4-v1.log", manifest);
+    verify_segment_request("1000-1-v1.log", manifest);
 
     BOOST_REQUIRE(part->archival_meta_stm());
     const cloud_storage::partition_manifest& stm_manifest
@@ -414,7 +414,7 @@ FIXTURE_TEST(test_upload_compacted_segments_concat, reupload_fixture) {
     BOOST_REQUIRE_EQUAL(replaced[1].base_offset, model::offset{1000});
 
     verify_concat_segment_request(
-      {"0-1-v1.log", "1000-4-v1.log"}, part->archival_meta_stm()->manifest());
+      {"0-1-v1.log", "1000-1-v1.log"}, part->archival_meta_stm()->manifest());
 }
 
 FIXTURE_TEST(
@@ -746,10 +746,10 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
     // NOTE: different terms so compaction leaves one segment each.
     std::vector<segment_desc> segments = {
       {manifest_ntp, model::offset(0), model::term_id(1), 10, 2},
-      {manifest_ntp, model::offset(10), model::term_id(2), 10, 2},
-      {manifest_ntp, model::offset(20), model::term_id(3), 10, 2},
-      {manifest_ntp, model::offset(30), model::term_id(4), 10, 2},
-      {manifest_ntp, model::offset(40), model::term_id(5), 10, 2},
+      {manifest_ntp, model::offset(10), model::term_id(1), 10, 2},
+      {manifest_ntp, model::offset(20), model::term_id(1), 10, 2},
+      {manifest_ntp, model::offset(30), model::term_id(1), 10, 2},
+      {manifest_ntp, model::offset(40), model::term_id(1), 10, 2},
     };
 
     initialize(segments);
@@ -768,9 +768,9 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
       get_targets().find(manifest_url)->second.content);
 
     verify_segment_request("0-1-v1.log", manifest);
-    verify_segment_request("10-2-v1.log", manifest);
-    verify_segment_request("20-3-v1.log", manifest);
-    verify_segment_request("30-4-v1.log", manifest);
+    verify_segment_request("10-1-v1.log", manifest);
+    verify_segment_request("20-1-v1.log", manifest);
+    verify_segment_request("30-1-v1.log", manifest);
 
     BOOST_REQUIRE(part->archival_meta_stm());
     const cloud_storage::partition_manifest& stm_manifest
@@ -790,7 +790,7 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
         create_segment(
           {manifest_ntp,
            last_segment->offsets().get_committed_offset() + model::offset{1},
-           model::term_id{6},
+           model::term_id{2},
            10});
     }
 
@@ -805,13 +805,13 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
     BOOST_REQUIRE_EQUAL(get_requests().size(), 9);
 
     verify_segment_request(
-      "40-5-v1.log", part->archival_meta_stm()->manifest());
+      "40-1-v1.log", part->archival_meta_stm()->manifest());
     verify_segment_request(
-      "50-6-v1.log", part->archival_meta_stm()->manifest());
+      "50-2-v1.log", part->archival_meta_stm()->manifest());
     verify_segment_request(
-      "65-6-v1.log", part->archival_meta_stm()->manifest());
+      "65-2-v1.log", part->archival_meta_stm()->manifest());
     verify_segment_request(
-      "85-6-v1.log", part->archival_meta_stm()->manifest());
+      "85-2-v1.log", part->archival_meta_stm()->manifest());
 
     BOOST_REQUIRE_EQUAL(
       stm_manifest.get_last_uploaded_compacted_offset(), model::offset{});
@@ -828,9 +828,9 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
     verify_concat_segment_request(
       {
         "0-1-v1.log",
-        "10-2-v1.log",
-        "20-3-v1.log",
-        "30-4-v1.log",
+        "10-1-v1.log",
+        "20-1-v1.log",
+        "30-1-v1.log",
       },
       part->archival_meta_stm()->manifest());
 
@@ -841,3 +841,74 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
     replaced = stm_manifest.replaced_segments();
     BOOST_REQUIRE_EQUAL(replaced.size(), 4);
 }
+
+FIXTURE_TEST(test_upload_compacted_segments_cross_term, reupload_fixture) {
+    std::vector<segment_desc> segments = {
+      {manifest_ntp, model::offset(0), model::term_id(1), 1000, 2},
+      {manifest_ntp, model::offset(1000), model::term_id(4), 10, 2},
+    };
+
+    initialize(segments);
+    auto action = ss::defer([this] { archiver->stop().get(); });
+
+    auto part = app.partition_manager.local().get(manifest_ntp);
+    listen();
+
+    // Upload two non compacted segments, no segment is compacted yet.
+    archival::ntp_archiver::batch_result expected{{2, 0, 0}, {0, 0, 0}};
+    upload_and_verify(archiver.value(), expected);
+
+    // Two segments, two indices, one manifest
+    BOOST_REQUIRE_EQUAL(get_requests().size(), 5);
+
+    auto manifest = verify_manifest_request(*part);
+    verify_segment_request("0-1-v1.log", manifest);
+    verify_segment_request("1000-4-v1.log", manifest);
+
+    BOOST_REQUIRE(part->archival_meta_stm());
+    const cloud_storage::partition_manifest& stm_manifest
+      = part->archival_meta_stm()->manifest();
+    verify_stm_manifest(stm_manifest, segments);
+
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get_last_uploaded_compacted_offset(), model::offset{});
+
+    // Mark both segments compacted, and re-upload. Both segments are
+    // re-uploaded.
+    reset_http_call_state();
+
+    vlog(test_log.info, "Waiting for segments to self-compact");
+    auto seg = self_compact_next_segment();
+    vlog(test_log.info, "Self-compaction completed");
+
+    expected = archival::ntp_archiver::batch_result{{0, 0, 0}, {2, 0, 0}};
+    upload_and_verify(archiver.value(), expected);
+    BOOST_REQUIRE_EQUAL(get_requests().size(), 5);
+
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get_last_uploaded_compacted_offset(),
+      seg->offsets().get_committed_offset());
+
+    auto replaced = stm_manifest.replaced_segments();
+    BOOST_REQUIRE_EQUAL(replaced.size(), 2);
+    BOOST_REQUIRE_EQUAL(replaced[0].base_offset, model::offset{0});
+    BOOST_REQUIRE_EQUAL(replaced[1].base_offset, model::offset{1000});
+
+    // We can't reupload x-term so we should end up with two
+    // compacted uploads.
+
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get(model::offset(0))->base_offset, model::offset(0));
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get(model::offset(0))->committed_offset, model::offset(999));
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get(model::offset(0))->segment_term, model::term_id(1));
+
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get(model::offset(1000))->base_offset, model::offset(1000));
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get(model::offset(1000))->committed_offset,
+      model::offset(1009));
+    BOOST_REQUIRE_EQUAL(
+      stm_manifest.get(model::offset(1000))->segment_term, model::term_id(4));
+}