storage: always schedule adjacent segment compaction

andrwng · andrwng · commit 1eb7c2d49e0c · 2025-02-24T15:08:25.000-08:00
CONFLICT: - this branch had code that takes a range instead of just the compaction config for compact_adjacent_segments() We previously fell back on adjacent segment compaction only if there was no new data to compact. In some situations, we've seen the rate of incoming data outpace the compaction interval, causing segments to pile up without ever being merged. This change tweaks the logic to always run adjacent segment compaction after running sliding window compaction. Along the way, a couple tests needed to be tweaked to handle the fact that housekeeping now may merge segments. (cherry picked from commit 08d0433)
diff --git a/src/v/archival/tests/ntp_archiver_reupload_test.cc b/src/v/archival/tests/ntp_archiver_reupload_test.cc
@@ -226,10 +226,11 @@ struct reupload_fixture : public archiver_fixture {
           manifest_view);
     }
 
+    // XXX: rename this! It doesn't just run self compaction -- it performs
+    // houskeeping which may also merge segments.
     ss::lw_shared_ptr<storage::segment> self_compact_next_segment(
       model::offset max_collectible = model::offset::max()) {
         auto& seg_set = disk_log_impl()->segments();
-        auto size_before = seg_set.size();
 
         disk_log_impl()
           ->housekeeping(storage::housekeeping_config{
@@ -240,13 +241,6 @@ struct reupload_fixture : public archiver_fixture {
             abort_source})
           .get();
 
-        auto size_after = seg_set.size();
-
-        // We are only looking to trigger self-compaction here.
-        // If the segment count reduced, adjacent segment compaction must
-        // have occurred.
-        BOOST_REQUIRE_EQUAL(size_before, size_after);
-
         ss::lw_shared_ptr<storage::segment> last_compacted_segment;
         for (auto& i : seg_set) {
             if (i->finished_self_compaction()) {
@@ -447,11 +441,11 @@ FIXTURE_TEST(
     std::stringstream st;
     stm_manifest.serialize_json(st);
     vlog(test_log.debug, "manifest: {}", st.str());
-    verify_segment_request("500-1-v1.log", stm_manifest);
 
     BOOST_REQUIRE_EQUAL(
       stm_manifest.get_last_uploaded_compacted_offset(), model::offset{999});
     auto replaced = stm_manifest.replaced_segments();
+    BOOST_REQUIRE_EQUAL(replaced.size(), 1);
     BOOST_REQUIRE_EQUAL(replaced[0].base_offset, model::offset{500});
 }
 
diff --git a/src/v/archival/tests/segment_reupload_test.cc b/src/v/archival/tests/segment_reupload_test.cc
@@ -832,6 +832,7 @@ SEASTAR_THREAD_TEST_CASE(test_upload_aligned_to_non_existent_offset) {
 
     auto first = spec.segment_starts.begin();
     auto second = std::next(first);
+    // Segment boundaries: [5, 14][15, 24][25, 34][35, 50]...
     for (; second != spec.segment_starts.end(); ++first, ++second) {
         b | storage::add_segment(*first);
         for (auto curr_offset = *first; curr_offset < *second; ++curr_offset) {
@@ -843,19 +844,18 @@ SEASTAR_THREAD_TEST_CASE(test_upload_aligned_to_non_existent_offset) {
               .get();
         }
         auto seg = b.get_log_segments().back();
-        seg->appender().close().get();
-        seg->release_appender().get();
+        seg->release_appender(&b.get_disk_log_impl().readers()).get();
     }
 
     b | storage::add_segment(*first)
       | storage::add_random_batch(*first, spec.last_segment_num_records);
 
-    // Compaction will rewrite each segment.
+    // Compaction will rewrite each segment, and merge the first two.
+    // Segment boundaries: [5, 24][25, 34][35, 50]...
     b.gc(model::timestamp::max(), std::nullopt).get();
 
     size_t max_size = b.get_segment(0).size_bytes()
-                      + b.get_segment(1).size_bytes()
-                      + b.get_segment(2).size_bytes();
+                      + b.get_segment(1).size_bytes();
     archival::segment_collector collector{
       model::offset{5}, m, b.get_disk_log_impl(), max_size};
 
@@ -868,6 +868,8 @@ SEASTAR_THREAD_TEST_CASE(test_upload_aligned_to_non_existent_offset) {
           ss::default_priority_class(), segment_lock_timeout)
         .get());
 
+    // The upload candidate should align with the manifest's segment
+    // boundaries.
     auto upload_candidate = upload_with_locks.candidate;
     BOOST_REQUIRE(!upload_candidate.sources.empty());
     BOOST_REQUIRE_EQUAL(upload_candidate.starting_offset, model::offset{10});
@@ -885,7 +887,7 @@ SEASTAR_THREAD_TEST_CASE(test_upload_aligned_to_non_existent_offset) {
     BOOST_REQUIRE_EQUAL(
       expected_content_length, upload_candidate.content_length);
 
-    BOOST_REQUIRE_EQUAL(upload_with_locks.read_locks.size(), 3);
+    BOOST_REQUIRE_EQUAL(upload_with_locks.read_locks.size(), 2);
 }
 
 SEASTAR_THREAD_TEST_CASE(test_same_size_reupload_skipped) {
diff --git a/src/v/storage/disk_log_impl.cc b/src/v/storage/disk_log_impl.cc
@@ -1243,24 +1243,28 @@ ss::future<> disk_log_impl::do_compact(
         std::rethrow_exception(eptr);
     }
     bool compacted = did_compact_fut.get();
-    if (!compacted) {
-        if (auto range = find_compaction_range(compact_cfg); range) {
-            auto r = co_await compact_adjacent_segments(
-              std::move(*range), compact_cfg);
-            vlog(
-              gclog.debug,
-              "Adjacent segments of {}, compaction result: {}",
-              config().ntp(),
-              r);
-            if (r.did_compact()) {
-                _compaction_ratio.update(r.compaction_ratio());
-            }
-        } else {
-            vlog(
-              gclog.debug,
-              "Adjacent segments of {}, no adjacent pair",
-              config().ntp());
+    vlog(
+      gclog.debug,
+      "Sliding compaction of {} did {}compact data, proceeding to adjacent "
+      "segment compaction",
+      config().ntp(),
+      compacted ? "" : "not ");
+    if (auto range = find_compaction_range(compact_cfg); range) {
+        auto r = co_await compact_adjacent_segments(
+          std::move(*range), compact_cfg);
+        vlog(
+          gclog.debug,
+          "Adjacent segments of {}, compaction result: {}",
+          config().ntp(),
+          r);
+        if (r.did_compact()) {
+            _compaction_ratio.update(r.compaction_ratio());
         }
+    } else {
+        vlog(
+          gclog.debug,
+          "Adjacent segments of {}, no adjacent pair",
+          config().ntp());
     }
 }
 
diff --git a/src/v/storage/tests/storage_e2e_test.cc b/src/v/storage/tests/storage_e2e_test.cc
@@ -1805,22 +1805,21 @@ FIXTURE_TEST(adjacent_segment_compaction, storage_test_fixture) {
 
     // There are 4 segments, and the last is the active segments. The first two
     // will merge, and the third will be compacted but not merged.
+
     log->housekeeping(c_cfg).get0();
-    BOOST_REQUIRE_EQUAL(log->segment_count(), 4);
+    BOOST_REQUIRE_EQUAL(log->segment_count(), 3);
 
     // Check if it honors max_compactible offset by resetting it to the base
     // offset of first segment. Nothing should be compacted.
     const auto first_segment_offsets = log->segments().front()->offsets();
     c_cfg.compact.max_collectible_offset
       = first_segment_offsets.get_base_offset();
     log->housekeeping(c_cfg).get0();
-    BOOST_REQUIRE_EQUAL(log->segment_count(), 4);
-
-    // The segment count will be reduced again.
-    c_cfg.compact.max_collectible_offset = model::offset::max();
-    log->housekeeping(c_cfg).get0();
     BOOST_REQUIRE_EQUAL(log->segment_count(), 3);
 
+    // Now compact without restricting collectible offset. The segment count
+    // will be reduced again.
+    c_cfg.compact.max_collectible_offset = model::offset::max();
     log->housekeeping(c_cfg).get0();
     BOOST_REQUIRE_EQUAL(log->segment_count(), 2);
 
@@ -1872,9 +1871,6 @@ FIXTURE_TEST(adjacent_segment_compaction_terms, storage_test_fixture) {
       as);
 
     // compact all the individual segments
-    log->housekeeping(c_cfg).get0();
-    BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 6);
-
     // the two segments with term 2 can be combined
     log->housekeeping(c_cfg).get0();
     BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 5);
@@ -1946,21 +1942,15 @@ FIXTURE_TEST(max_adjacent_segment_compaction, storage_test_fixture) {
       as);
 
     // self compaction steps
-    log->housekeeping(c_cfg).get0();
-    BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 6);
-
     // the first two segments are combined 2+2=4 < 6 MB
     log->housekeeping(c_cfg).get0();
     BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 5);
 
     // the new first and second are too big 4+5 > 6 MB but the second and third
     // can be combined 5 + 15KB < 6 MB
-    log->housekeeping(c_cfg).get0();
-    BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 4);
-
     // then the next 16 KB can be folded in
     log->housekeeping(c_cfg).get0();
-    BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 3);
+    BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 4);
 
     // that's all that can be done. the next seg is an appender
     log->housekeeping(c_cfg).get0();
@@ -2159,16 +2149,16 @@ FIXTURE_TEST(compaction_backlog_calculation, storage_test_fixture) {
     // self compaction steps
     log->housekeeping(c_cfg).get0();
 
-    BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 5);
+    BOOST_REQUIRE_EQUAL(disk_log->segment_count(), 4);
     auto new_backlog_size = log->compaction_backlog();
     /**
      * after all self segments are compacted they shouldn't be included into the
      * backlog (only last segment is since it has appender and isn't self
      * compacted)
      */
-    BOOST_REQUIRE_EQUAL(
+    BOOST_REQUIRE_LT(
       new_backlog_size,
-      backlog_size - self_seg_compaction_sz + segments[4]->size_bytes());
+      backlog_size - self_seg_compaction_sz + segments[3]->size_bytes());
 }
 
 FIXTURE_TEST(not_compacted_log_backlog, storage_test_fixture) {
