From 5ff59465206d201cc6f2157745c84042d21e9c25 Mon Sep 17 00:00:00 2001
From: Pavel Emelyanov <xemul@scylladb.com>
Date: Thu, 30 Nov 2023 16:36:32 +0300
Subject: [PATCH] sstables: Split remove_by_toc_name()

The helper consists of three phases:

- move TOC -> TOC.tmp
- remove components listed in TOC
- remove TOC.tmp

The first step is needed separately by the next patch

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>
---
 sstables/sstables.cc | 15 +++++++++++++--
 sstables/sstables.hh |  5 +++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/sstables/sstables.cc b/sstables/sstables.cc
index deac9e4c80..be0cd53465 100644
--- a/sstables/sstables.cc
+++ b/sstables/sstables.cc
@@ -2601,7 +2601,7 @@ std::optional<std::pair<uint64_t, uint64_t>> sstable::get_sample_indexes_for_ran
     return std::nullopt;
 }
 
-future<> remove_by_toc_name(sstring sstable_toc_name, storage::sync_dir sync) {
+future<sstring> make_toc_temporary(sstring sstable_toc_name, storage::sync_dir sync) {
     sstring prefix = sstable_toc_name.substr(0, sstable_toc_name.size() - sstable_version_constants::TOC_SUFFIX.size());
     sstring new_toc_name = prefix + sstable_version_constants::TEMPORARY_TOC_SUFFIX;
 
@@ -2615,9 +2615,20 @@ future<> remove_by_toc_name(sstring sstable_toc_name, storage::sync_dir sync) {
     } else {
         if (!co_await sstable_io_check(sstable_write_error_handler, file_exists, new_toc_name)) {
             sstlog.warn("Unable to delete {} because it doesn't exist.", sstable_toc_name);
-            co_return;
+            co_return "";
         }
     }
+    co_return new_toc_name;
+}
+
+future<> remove_by_toc_name(sstring sstable_toc_name, storage::sync_dir sync) {
+    sstring prefix = sstable_toc_name.substr(0, sstable_toc_name.size() - sstable_version_constants::TOC_SUFFIX.size());
+
+    auto new_toc_name = co_await make_toc_temporary(sstable_toc_name, sync);
+    if (new_toc_name.empty()) {
+        co_return;
+    }
+
     auto toc_file = co_await open_checked_file_dma(sstable_write_error_handler, new_toc_name, open_flags::ro);
     std::vector<sstring> components = co_await with_closeable(std::move(toc_file), [] (file& toc_file) {
         return sstable::read_and_parse_toc(toc_file);
diff --git a/sstables/sstables.hh b/sstables/sstables.hh
index 2991c83f6d..7d67891b4b 100644
--- a/sstables/sstables.hh
+++ b/sstables/sstables.hh
@@ -1021,5 +1021,10 @@ future<> remove_table_directory_if_has_no_snapshots(fs::path table_dir);
 // Caller may pass sync_dir::no for batching multiple deletes in the same directory,
 // and make sure the directory is sync'ed on or after the last call.
 future<> remove_by_toc_name(sstring sstable_toc_name, storage::sync_dir sync = storage::sync_dir::yes);
+// makes sure the TOC file is temporary by moving existing TOC file or otherwise
+// checking the temporary-TOC already exists
+// resolves into temporary-TOC file name or empty string if neither TOC nor temp.
+// TOC is there
+future<sstring> make_toc_temporary(sstring sstable_toc_name, storage::sync_dir sync = storage::sync_dir::yes);
 
 } // namespace sstables