storage: always schedule adjacent segment compaction

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)

Author: andrwng

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

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) {

src/v/storage/disk_log_impl.cc

@@ -1243,7 +1243,12 @@ ss::future<> disk_log_impl::do_compact(
         std::rethrow_exception(eptr);
     }
     bool compacted = did_compact_fut.get();
-    if (!compacted) {
+    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);
@@ -1261,7 +1266,6 @@ ss::future<> disk_log_impl::do_compact(
               "Adjacent segments of {}, no adjacent pair",
               config().ntp());
         }
-    }
 }
 
 ss::future<> disk_log_impl::gc(gc_config cfg) {

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) {