From bde18c4e518ef12d83697563b59a5cecee2bbeed Mon Sep 17 00:00:00 2001
From: Michael Litvak <michael.litvak@scylladb.com>
Date: Sun, 24 May 2026 10:05:14 +0200
Subject: [PATCH] logstor: compaction state cleanup

add a simple cleanup for the logstor compaction state map to remove
entries of stale compaction groups. remove the state of compaction group
from the map if it doesn't have anything in progress.
---
 replica/logstor/segment_manager.cc | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/replica/logstor/segment_manager.cc b/replica/logstor/segment_manager.cc
index a45b93a3a9..da959b574d 100644
--- a/replica/logstor/segment_manager.cc
+++ b/replica/logstor/segment_manager.cc
@@ -410,8 +410,13 @@ private:
         shared_future<> completion{make_ready_future<>()};
         abort_source as;
         int compaction_disabled_counter{0};
+
+        bool is_empty() const noexcept {
+            return !running && compaction_disabled_counter == 0 && completion.available();
+        }
     };
-    absl::flat_hash_map<compaction_group*, std::unique_ptr<group_compaction_state>> _groups;
+    using group_compaction_state_map = absl::flat_hash_map<compaction_group*, std::unique_ptr<group_compaction_state>>;
+    group_compaction_state_map _groups;
 
 public:
     compaction_manager_impl(segment_manager_impl& sm, compaction_config cfg)
@@ -440,6 +445,12 @@ public:
 
 private:
 
+    void maybe_erase_state(group_compaction_state_map::iterator it) {
+        if (it->second->is_empty()) {
+            _groups.erase(it);
+        }
+    }
+
     std::vector<log_segment_id> select_segments_for_compaction(const segment_descriptor_hist&);
     future<> do_compact(compaction_group&, abort_source&);
     future<> compact_segments(compaction_group&, std::vector<log_segment_id>);
@@ -1426,6 +1437,7 @@ future<compaction_reenabler> compaction_manager_impl::disable_compaction(compact
         auto it = _groups.find(&cg);
         if (it != _groups.end()) {
             --it->second->compaction_disabled_counter;
+            maybe_erase_state(it);
         }
     });
 }
@@ -1443,6 +1455,7 @@ compaction_reenabler compaction_manager_impl::disable_compaction_no_wait(compact
         auto it = _groups.find(&cg);
         if (it != _groups.end()) {
             --it->second->compaction_disabled_counter;
+            maybe_erase_state(it);
         }
     });
 }