[#273] Fix race condition in migration scheduler

The lambda function fed to the irods::query_processor for
scheduling violating data objects for migration had a race
condition wherein objects marked as already having been
processed were not protected by a mutex. This resulted in
data objects with multiple users or groups with permissions
on those data objects being scheduled for migration multiple
times.

src/storage_tiering.cpp

@@ -33,6 +33,7 @@
 #include <nlohmann/json.hpp>
 
 #include <charconv>
+#include <mutex>
 #include <random>
 #include <system_error>
 #include <tuple>
@@ -576,7 +577,9 @@ namespace irods {
 
         irods::thread_pool thread_pool{config_.number_of_scheduling_threads};
         try {
+            // TODO(#298): Consider changing this from std::map to std::unordered_set since the value is never used.
             std::map<std::string, uint8_t> object_is_processed;
+            std::mutex object_is_processed_mutex;
             const bool preserve_replicas = get_preserve_replicas_for_resc(_comm, _source_resource);
             const auto query_limit       = get_object_limit_for_resource(_comm, _source_resource);
             const auto query_list        = get_violating_queries_for_resource(_comm, _source_resource);
@@ -617,12 +620,18 @@ namespace irods {
                     }
                     object_path += _results[0]; // data name
 
-                    if(std::end(object_is_processed) !=
-                       object_is_processed.find(object_path)) {
-                        return;
-                    }
+                    {
+                        // An irods::query_processor concurrently executes this function for each returned result. Each
+                        // instance refers to the same object_is_processed instance. So, we need a lock here to protect
+                        // against concurrent accesses of the object_is_processed map.
+                        const std::lock_guard object_is_processed_lock{object_is_processed_mutex};
+
+                        if (std::end(object_is_processed) != object_is_processed.find(object_path)) {
+                            return;
+                        }
 
-                    object_is_processed[object_path] = 1;
+                        object_is_processed[object_path] = 1;
+                    }
 
                     auto proxy_conn = irods::proxy_connection();
                     rcComm_t* comm = proxy_conn.make_rodsadmin_connection();