cloud_storage: Force exhaustive trim when fast trim fails

In the case where the cache dir consists solely of index+tx files, the
current code path in fast trim does not remove anything. This should be
followed up by an exhaustive trim to free up slots in cache, but due to
an adjustment of the objects to delete counter, this ends up not
happening.

The change here makes sure that if we have a certain count of objects to
delete, and we were not able to delete the count, and there is a set of
filtered out files, we proceed to do an exhaustive trim.

src/v/cloud_storage/cache_service.cc

@@ -351,7 +351,8 @@ ss::future<> cache::trim(
     vlog(
       cst_log.debug,
       "trim: set target_size {}/{}, size {}/{}, walked size {} (max {}/{}), "
-      " reserved {}/{}, pending {}/{})",
+      " reserved {}/{}, pending {}/{}), candidates for deletion: {}, filtered "
+      "out: {}",
       target_size,
       target_objects,
       _current_cache_size,
@@ -362,7 +363,9 @@ ss::future<> cache::trim(
       _reserved_cache_size,
       _reserved_cache_objects,
       _reservations_pending,
-      _reservations_pending_objects);
+      _reservations_pending_objects,
+      candidates_for_deletion.size(),
+      filtered_out_files);
 
     if (
       _current_cache_size + _reserved_cache_size < target_size
@@ -453,9 +456,24 @@ ss::future<> cache::trim(
     // cache.
     size_to_delete = std::min(
       walked_cache_size - fast_result.deleted_size, size_to_delete);
-    objects_to_delete = std::min(
-      candidates_for_deletion.size() - fast_result.deleted_count,
-      objects_to_delete);
+
+    // If we were not able to delete enough files and there are some filtered
+    // out files, force an exhaustive trim. This ensures that if the cache is
+    // dominated by filtered out files, we do not skip trimming them by reducing
+    // the objects_to_delete counter next.
+    bool force_exhaustive_trim = fast_result.deleted_count < objects_to_delete
+                                 && filtered_out_files > 0;
+
+    // In the situation where all files in cache are filtered out,
+    // candidates_for_deletion equals 1 (due to the accesstime tracker file) and
+    // the following reduction to objects_to_delete ends up setting
+    // this counter to 1, causing the exhaustive trim to be skipped. The check
+    // force_exhaustive_trim avoids this.
+    if (!force_exhaustive_trim) {
+        objects_to_delete = std::min(
+          candidates_for_deletion.size() - fast_result.deleted_count,
+          objects_to_delete);
+    }
 
     if (
       size_to_delete > undeletable_bytes

src/v/cloud_storage/tests/cache_test.cc

@@ -26,6 +26,7 @@
 #include <boost/test/unit_test.hpp>
 
 #include <chrono>
+#include <fstream>
 #include <stdexcept>
 
 using namespace cloud_storage;
@@ -466,3 +467,36 @@ FIXTURE_TEST(test_clean_up_on_start_empty, cache_test_fixture) {
 
     BOOST_CHECK(ss::file_exists(CACHE_DIR.native()).get());
 }
+
+/**
+ * Given a cache dir populated with files which are filtered out by fast trim,
+ * validate that a failing fast trim should be followed up by an exhaustive trim
+ * and clean up the required object count.
+ */
+FIXTURE_TEST(test_exhaustive_trim_runs_after_fast_trim, cache_test_fixture) {
+    std::vector<std::filesystem::path> indices;
+    const auto count_indices = 5;
+    indices.reserve(count_indices);
+
+    for (auto i = 0; i < count_indices; ++i) {
+        indices.emplace_back(CACHE_DIR / fmt::format("{}.index", i, i, i));
+        std::ofstream f{indices.back()};
+        f.flush();
+    }
+
+    BOOST_REQUIRE(
+      std::all_of(indices.cbegin(), indices.cend(), [](const auto& path) {
+          return std::filesystem::exists(path);
+      }));
+
+    // Make cache service scan the disk for objects
+    clean_up_at_start().get();
+
+    // Only allow the access time tracker to remain on disk.
+    trim_cache(std::nullopt, 1);
+
+    BOOST_REQUIRE(
+      std::all_of(indices.cbegin(), indices.cend(), [](const auto& path) {
+          return !std::filesystem::exists(path);
+      }));
+}

src/v/cloud_storage/tests/cache_test_fixture.h

@@ -107,10 +107,16 @@ class cache_test_fixture {
         return sharded_cache.local().clean_up_at_start();
     }
 
-    void trim_cache() {
+    void trim_cache(
+      std::optional<uint64_t> size_limit_override = std::nullopt,
+      std::optional<size_t> object_limit_override = std::nullopt) {
         sharded_cache
           .invoke_on(
-            ss::shard_id{0}, [](cloud_storage::cache& c) { return c.trim(); })
+            ss::shard_id{0},
+            [&size_limit_override,
+             &object_limit_override](cloud_storage::cache& c) {
+                return c.trim(size_limit_override, object_limit_override);
+            })
           .get();
     }
 };