Merge 'Skip mode validation for snapshots' from Benny Halevy

Skip over verification of owner and mode of the snapshots
sub-directory as this might race with scylla-manager
trying to delete old snapshots concurrently.

Fixes #12010

Closes #14892

* github.com:scylladb/scylladb:
  distributed_loader: process_sstable_dir: do not verify snapshots
  utils/directories: verify_owner_and_mode: add recursive flag

replica/distributed_loader.cc

@@ -85,8 +85,18 @@ io_error_handler error_handler_gen_for_upload_dir(disk_error_signal_type& dummy)
 
 future<>
 distributed_loader::process_sstable_dir(sharded<sstables::sstable_directory>& dir, sstables::sstable_directory::process_flags flags) {
-    co_await dir.invoke_on(0, [] (const sstables::sstable_directory& d) {
-        return utils::directories::verify_owner_and_mode(d.sstable_dir());
+    // verify owner and mode on the sstables directory
+    // and all its subdirectories, except for "snapshots"
+    // as there could be a race with scylla-manager that might
+    // delete snapshots concurrently
+    co_await dir.invoke_on(0, [] (const sstables::sstable_directory& d) -> future<> {
+        fs::path sstable_dir = d.sstable_dir();
+        co_await utils::directories::verify_owner_and_mode(sstable_dir, utils::directories::recursive::no);
+        co_await lister::scan_dir(sstable_dir, lister::dir_entry_types::of<directory_entry_type::directory>(), [] (fs::path dir, directory_entry de) -> future<> {
+            if (de.name != sstables::snapshots_dir) {
+                co_await utils::directories::verify_owner_and_mode(dir / de.name, utils::directories::recursive::yes);
+            }
+        });
     });
 
     if (flags.garbage_collect) {

utils/directories.cc

@@ -106,7 +106,7 @@ void verification_error(fs::path path, const char* fstr, Args&&... args) {
 // Verify that all files and directories are owned by current uid
 // and that files can be read and directories can be read, written, and looked up (execute)
 // No other file types may exist.
-future<> directories::verify_owner_and_mode(fs::path path) {
+future<> directories::do_verify_owner_and_mode(fs::path path, recursive recurse, int level) {
     auto sd = co_await file_stat(path.string(), follow_symlink::no);
     // Under docker, we run with euid 0 and there is no reasonable way to enforce that the
     // in-container uid will have the same uid as files mounted from outside the container. So
@@ -128,8 +128,11 @@ future<> directories::verify_owner_and_mode(fs::path path) {
         if (!can_access) {
             verification_error(std::move(path), "Directory cannot be accessed for read, write, and execute");
         }
-        co_await lister::scan_dir(path, {}, [] (fs::path dir, directory_entry de) -> future<> {
-            co_await verify_owner_and_mode(dir / de.name);
+        if (level && !recurse) {
+            co_return;
+        }
+        co_await lister::scan_dir(path, {}, [recurse, level = level + 1] (fs::path dir, directory_entry de) -> future<> {
+            co_await do_verify_owner_and_mode(dir / de.name, recurse, level);
         });
         break;
     }
@@ -138,4 +141,8 @@ future<> directories::verify_owner_and_mode(fs::path path) {
     }
 };
 
+future<> directories::verify_owner_and_mode(fs::path path, recursive recursive) {
+    return do_verify_owner_and_mode(std::move(path), recursive, 0);
+}
+
 } // namespace utils

utils/directories.hh

@@ -39,12 +39,16 @@ public:
         std::set<fs::path> _paths;
     };
 
+    using recursive = bool_class<struct recursive_tag>;
+
     directories(bool developer_mode);
     future<> create_and_verify(set dir_set);
-    static future<> verify_owner_and_mode(std::filesystem::path path);
+    static future<> verify_owner_and_mode(std::filesystem::path path, recursive r = recursive::yes);
 private:
     bool _developer_mode;
     std::vector<file_lock> _locks;
+
+    static future<> do_verify_owner_and_mode(std::filesystem::path path, recursive, int level);
 };
 
 } // namespace utils