ct/l0/gc: Each iteration should page through list results

Limited by the the semaphore in list_delete_worker.

Signed-off-by: Oren Leiman <oren.leiman@redpanda.com>

Author: oleiman

src/v/cloud_topics/level_zero/gc/level_zero_gc.cc

@@ -604,9 +604,27 @@ seastar::future<> level_zero_gc::worker() {
 
 seastar::future<std::expected<size_t, level_zero_gc::collection_error>>
 level_zero_gc::try_to_collect() {
-    if (!delete_worker_->has_capacity()) {
-        co_return 0;
+    // Ultra-temporary cache to avoid repeatedly querying for max gc-able epoch.
+    // Since the result will always be valid clusterwide, compute exactly once
+    // per collection loop.
+    std::optional<cluster_epoch> max_gc_epoch;
+    size_t total_eligible{0};
+    while (delete_worker_->has_capacity()) {
+        auto res = co_await do_try_to_collect(std::ref(max_gc_epoch));
+        if (!res.has_value()) {
+            co_return res;
+        }
+        if (res.value() == 0) {
+            break;
+        }
+        total_eligible += res.value();
     }
+
+    co_return total_eligible;
+}
+
+seastar::future<std::expected<size_t, level_zero_gc::collection_error>>
+level_zero_gc::do_try_to_collect(std::optional<cluster_epoch>& max_gc_epoch) {
     auto candidate_objects = co_await delete_worker_->next_page();
     if (!candidate_objects.has_value()) {
         vlog(
@@ -616,17 +634,19 @@ level_zero_gc::try_to_collect() {
         co_return std::unexpected(collection_error::service_error);
     }
 
-    const auto maybe_max_gc_epoch
-      = co_await epoch_source_->max_gc_eligible_epoch(&asrc_);
-    if (!maybe_max_gc_epoch.has_value()) {
-        vlog(
-          cd_log.debug,
-          "Received error retrieving GC eligible epoch: {}",
-          maybe_max_gc_epoch.error());
-        co_return std::unexpected(collection_error::service_error);
+    if (!max_gc_epoch.has_value()) {
+        const auto maybe_max_gc_epoch
+          = co_await epoch_source_->max_gc_eligible_epoch(&asrc_);
+        if (!maybe_max_gc_epoch.has_value()) {
+            vlog(
+              cd_log.debug,
+              "Received error retrieving GC eligible epoch: {}",
+              maybe_max_gc_epoch.error());
+            co_return std::unexpected(collection_error::service_error);
+        }
+        max_gc_epoch = maybe_max_gc_epoch.value();
     }
 
-    const auto max_gc_epoch = maybe_max_gc_epoch.value();
     if (!max_gc_epoch.has_value()) {
         vlog(cd_log.info, "No GC eligible epoch currently exists");
         co_return std::unexpected(collection_error::no_collectible_epoch);

src/v/cloud_topics/level_zero/gc/level_zero_gc.h

@@ -275,6 +275,8 @@ class level_zero_gc {
     seastar::future<> worker();
     enum class collection_error : int8_t;
     seastar::future<std::expected<size_t, collection_error>> try_to_collect();
+    seastar::future<std::expected<size_t, collection_error>>
+    do_try_to_collect(std::optional<cluster_epoch>&);
 
     level_zero_gc_probe probe_;
 

src/v/cloud_topics/level_zero/gc/tests/level_zero_gc_tests.cc

@@ -390,3 +390,34 @@ TEST_F(LevelZeroGCScaleOutTest, CleanShutdown) {
     // then immediately shutdown gc
     gc.stop().get();
 }
+
+TEST_F(LevelZeroGCScaleOutTest, ConcurrentDeletes) {
+    static constexpr size_t list_page_size{20};
+    int n = list_page_size * 20;
+    for (int i = 0; i < n; i++) {
+        add_listed(i, 24h);
+    }
+    this->max_epoch = n;
+    this->cfg.list_page_size = list_page_size;
+    this->cfg.delete_cost = 100ms;
+    gc.start();
+    EXPECT_TRUE(Eventually(
+      [this, expected = (size_t)n] { return deleted.size() == expected; }));
+}
+
+// make sure we make progress when the total size of eligible keys exceeds the
+// in-flight limit
+// i.e. 50 * 500 * len(key)=72 ~= 1.5MiB > 1MiB
+TEST_F(LevelZeroGCScaleOutTest, ConcurrentDeletesPipelineSaturation) {
+    static constexpr size_t list_page_size{500};
+    int n = list_page_size * 50;
+    for (int i = 0; i < n; i++) {
+        add_listed(i, 24h);
+    }
+    this->max_epoch = n;
+    this->cfg.list_page_size = list_page_size;
+    this->cfg.delete_cost = 50ms;
+    gc.start();
+    EXPECT_TRUE(Eventually(
+      [this, expected = (size_t)n] { return deleted.size() == expected; }));
+}