cloud_storage: add unit tests for cleanup on startup

...including the new empty dir deletion.

src/v/cloud_storage/cache_service.h

@@ -33,6 +33,8 @@ static constexpr size_t default_write_buffer_size = 128_KiB;
 static constexpr unsigned default_writebehind = 10;
 static constexpr const char* access_time_tracker_file_name = "accesstime";
 
+class cache_test_fixture;
+
 struct cache_item {
     ss::file body;
     size_t size;
@@ -130,6 +132,8 @@ class cache : public ss::peering_sharded_service<cache> {
     cache_probe probe;
     access_time_tracker _access_time_tracker;
     ss::timer<ss::lowres_clock> _tracker_timer;
+
+    friend class cache_test_fixture;
 };
 
 } // namespace cloud_storage

src/v/cloud_storage/tests/cache_test.cc

@@ -14,6 +14,7 @@
 #include "cloud_storage/cache_service.h"
 #include "test_utils/fixture.h"
 #include "units.h"
+#include "utils/file_io.h"
 
 #include <seastar/core/fstream.hh>
 #include <seastar/core/seastar.hh>
@@ -369,3 +370,46 @@ SEASTAR_THREAD_TEST_CASE(test_access_time_tracker_serializer) {
         BOOST_REQUIRE(ts.value() >= timestamps[i]);
     }
 }
+
+/**
+ * Validate that .part files and empty directories are deleted if found during
+ * the startup walk of the cache.
+ */
+FIXTURE_TEST(test_clean_up_on_start, cache_test_fixture) {
+    // A temporary file, this should be deleted on startup
+    put_into_cache(create_data_string('a', 1_KiB), KEY);
+    std::filesystem::path tmp_key = KEY;
+    tmp_key.replace_extension(".part");
+    ss::rename_file((CACHE_DIR / KEY).native(), (CACHE_DIR / tmp_key).native())
+      .get();
+
+    // A normal looking segment, we'll check this isn't deleted
+    put_into_cache(create_data_string('b', 1_KiB), KEY);
+
+    // An empty directory, this should be deleted
+    auto empty_dir_path = std::filesystem::path{CACHE_DIR / "empty_dir"};
+    ss::make_directory(empty_dir_path.native()).get();
+
+    // A non-empty-directory, this should be preserved
+    auto populated_dir_path = CACHE_DIR / "populated_dir";
+    ss::make_directory(populated_dir_path.native()).get();
+    write_fully(populated_dir_path / "populated_file", {}).get();
+
+    clean_up_at_start().get();
+
+    BOOST_CHECK(ss::file_exists((CACHE_DIR / KEY).native()).get());
+    BOOST_CHECK(!ss::file_exists((CACHE_DIR / tmp_key).native()).get());
+    BOOST_CHECK(!ss::file_exists(empty_dir_path.native()).get());
+    BOOST_CHECK(ss::file_exists(populated_dir_path.native()).get());
+}
+
+/**
+ * Validate that the empty directory deletion code in clean_up_at_start
+ * does not consider the cache directory itself an empty directory
+ * to be deleted.
+ */
+FIXTURE_TEST(test_clean_up_on_start_empty, cache_test_fixture) {
+    clean_up_at_start().get();
+
+    BOOST_CHECK(ss::file_exists(CACHE_DIR));
+}

src/v/cloud_storage/tests/cache_test_fixture.h

@@ -25,13 +25,16 @@
 #include <chrono>
 #include <filesystem>
 
-using namespace cloud_storage;
 using namespace std::chrono_literals;
 
 static inline std::filesystem::path get_cache_dir(std::filesystem::path p) {
     return p / "test_cache_dir";
 }
 
+// In cloud_storage namespace so we can befriend this fixture from
+// the class under test.
+namespace cloud_storage {
+
 class cache_test_fixture {
 public:
     const std::filesystem::path KEY{"abc001/test_topic/test_cache_file.txt"};
@@ -72,4 +75,10 @@ class cache_test_fixture {
         auto input = make_iobuf_input_stream(std::move(buf));
         sharded_cache.local().put(key, input).get();
     }
+
+    ss::future<> clean_up_at_start() {
+        return sharded_cache.local().clean_up_at_start();
+    }
 };
+
+} // namespace cloud_storage