sstables: close_files: auto-remove temporary sstable directory

If the sstable is marked for deletion, e.g. when
writing the sstable fails for any reason before it's
sealed, make sure to remove the sstable's temporary
directory, if present, besides the sstables files.

This condition is benign as these empty temp dirs
are removed when scylla starts up, but the do accumulate
and we better remove them too.

Fixes #9522

Test: unit(dev)

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
Message-Id: <20220302161827.2448980-1-bhalevy@scylladb.com>

sstables/sstables.cc

@@ -21,6 +21,8 @@
 #include <seastar/core/byteorder.hh>
 #include <seastar/core/aligned_buffer.hh>
 #include <seastar/core/metrics.hh>
+#include <seastar/core/reactor.hh>
+#include <seastar/util/file.hh>
 #include <seastar/util/closeable.hh>
 #include <iterator>
 #include <seastar/core/coroutine.hh>
@@ -2462,6 +2464,7 @@ future<> sstable::close_files() {
     }
 
     auto unlinked = make_ready_future<>();
+    auto unlinked_temp_dir = make_ready_future<>();
     if (_marked_for_deletion != mark_for_deletion::none) {
         // If a deletion fails for some reason we
         // log and ignore this failure, because on startup we'll again try to
@@ -2481,11 +2484,24 @@ future<> sstable::close_files() {
             sstlog.warn("Exception when deleting sstable file: {}", std::current_exception());
         }
 
+        if (_temp_dir) {
+            try {
+                unlinked_temp_dir = recursive_remove_directory(fs::path(*_temp_dir)).then_wrapped([this] (future<> f) {
+                    if (f.failed()) {
+                        sstlog.warn("Exception when deleting temporary sstable directory {}: {}", *_temp_dir, f.get_exception());
+                    } else {
+                        _temp_dir.reset();
+                    }
+                });
+            } catch (...) {
+                sstlog.warn("Exception when deleting temporary sstable directory {}: {}", *_temp_dir, std::current_exception());
+            }
+        }
     }
 
     _on_closed(*this);
 
-    return when_all_succeed(std::move(index_closed), std::move(data_closed), std::move(unlinked)).discard_result().then([this] {
+    return when_all_succeed(std::move(index_closed), std::move(data_closed), std::move(unlinked), std::move(unlinked_temp_dir)).discard_result().then([this, me = shared_from_this()] {
         if (_open_mode) {
             if (_open_mode.value() == open_flags::ro) {
                 _stats.on_close_for_reading();