cst/cache: Test current exception handling during put

src/v/cloud_storage/tests/BUILD

@@ -270,8 +270,10 @@ redpanda_cc_btest(
         "//src/v/cloud_storage",
         "//src/v/config",
         "//src/v/random:generators",
+        "//src/v/test_utils:iostream",
         "//src/v/test_utils:scoped_config",
         "//src/v/test_utils:seastar_boost",
+        "//src/v/utils:directory_walker",
         "//src/v/utils:file_io",
         "//src/v/utils:human",
         "@boost//:filesystem",

src/v/cloud_storage/tests/cache_test.cc

@@ -16,7 +16,9 @@
 #include "cloud_storage/cache_service.h"
 #include "random/generators.h"
 #include "test_utils/fixture.h"
+#include "test_utils/iostream.h"
 #include "test_utils/scoped_config.h"
+#include "utils/directory_walker.h"
 #include "utils/file_io.h"
 #include "utils/human.h"
 
@@ -465,6 +467,43 @@ SEASTAR_THREAD_TEST_CASE(test_access_time_tracker_read_skipped_on_old_version) {
     BOOST_REQUIRE_EQUAL(out.size(), 0);
 }
 
+ss::future<size_t> count_files(ss::sstring dirname) {
+    directory_walker walker;
+    size_t count = 0;
+    co_await walker.walk(
+      dirname,
+      [dirname, &count](const ss::directory_entry& entry) -> ss::future<> {
+          if (entry.type == ss::directory_entry_type::directory) {
+              return count_files(fmt::format("{}/{}", dirname, entry.name))
+                .then([&count](size_t sub_count) { count += sub_count; });
+          } else {
+              ++count;
+              return ss::now();
+          }
+      });
+
+    co_return count;
+}
+
+FIXTURE_TEST(test_clean_up_on_stream_exception, cache_test_fixture) {
+    auto s = tests::make_throwing_stream(ss::abort_requested_exception());
+    auto reservation = sharded_cache.local().reserve_space(1, 1).get();
+    BOOST_CHECK_THROW(
+      sharded_cache.local().put(KEY, s, reservation).get(),
+      ss::abort_requested_exception);
+    vlog(test_log.info, "Put failed as expected");
+
+    BOOST_CHECK_EQUAL(sharded_cache.local().get_usage_bytes(), 0);
+    BOOST_CHECK_EQUAL(sharded_cache.local().get_usage_objects(), 0);
+
+    // TODO: This is not expected behavior. The temporary file should be cleaned
+    // up on exception.
+    vlog(test_log.info, "Counting files in cache directory");
+    BOOST_CHECK_EQUAL(count_files(CACHE_DIR.native()).get(), 1);
+
+    vlog(test_log.info, "Test passed");
+}
+
 /**
  * Validate that .part files and empty directories are deleted if found during
  * the startup walk of the cache.

src/v/utils/tests/BUILD

@@ -402,8 +402,8 @@ redpanda_cc_btest(
         "//src/v/bytes:iostream",
         "//src/v/bytes:random",
         "//src/v/random:generators",
-        "//src/v/test_utils:seastar_boost",
         "//src/v/test_utils:iostream",
+        "//src/v/test_utils:seastar_boost",
         "//src/v/utils:stream_utils",
         "@boost//:test",
         "@seastar",