From 675a2fc40224df895387e115960d34eef8e72f33 Mon Sep 17 00:00:00 2001
From: Kishore Nallan <kishorenc@gmail.com>
Date: Sat, 16 Sep 2023 20:45:09 +0530
Subject: [PATCH] Prevent multiple alters from running in-parallel.

---
 include/core_api.h  |  4 ++++
 src/core_api.cpp    | 15 ++++++++++++++-
 src/raft_server.cpp | 14 ++++++++++++++
 3 files changed, 32 insertions(+), 1 deletion(-)

diff --git a/include/core_api.h b/include/core_api.h
index 6d4fe6a9..85b5469b 100644
--- a/include/core_api.h
+++ b/include/core_api.h
@@ -9,6 +9,10 @@ bool handle_authentication(std::map<std::string, std::string>& req_params,
                            const std::string& body, const route_path& rpath,
                            const std::string& req_auth_key);
 
+void set_alter_in_progress(bool in_progress);
+
+bool get_alter_in_progress();
+
 // Collections
 
 bool get_collections(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res);
diff --git a/src/core_api.cpp b/src/core_api.cpp
index e82b5963..96749bcf 100644
--- a/src/core_api.cpp
+++ b/src/core_api.cpp
@@ -1,6 +1,5 @@
 #include <chrono>
 #include <thread>
-#include <cstdlib> 
 #include <app_metrics.h>
 #include <regex>
 #include <analytics_manager.h>
@@ -23,10 +22,20 @@ using namespace std::chrono_literals;
 std::shared_mutex mutex;
 LRU::Cache<uint64_t, cached_res_t> res_cache;
 
+std::atomic<bool> alter_in_progress = false;
+
 void init_api(uint32_t cache_num_entries) {
     res_cache.capacity(cache_num_entries);
 }
 
+void set_alter_in_progress(bool in_progress) {
+    alter_in_progress = in_progress;
+}
+
+bool get_alter_in_progress() {
+    return alter_in_progress;
+}
+
 bool handle_authentication(std::map<std::string, std::string>& req_params,
                            std::vector<nlohmann::json>& embedded_params_vec,
                            const std::string& body,
@@ -218,6 +227,7 @@ bool patch_update_collection(const std::shared_ptr<http_req>& req, const std::sh
     } catch(const std::exception& e) {
         //LOG(ERROR) << "JSON error: " << e.what();
         res->set_400("Bad JSON.");
+        alter_in_progress = false;
         return false;
     }
 
@@ -226,15 +236,18 @@ bool patch_update_collection(const std::shared_ptr<http_req>& req, const std::sh
 
     if(collection == nullptr) {
         res->set_404();
+        alter_in_progress = false;
         return false;
     }
 
     auto alter_op = collection->alter(req_json);
     if(!alter_op.ok()) {
         res->set(alter_op.code(), alter_op.error());
+        alter_in_progress = false;
         return false;
     }
 
+    alter_in_progress = false;
     res->set_200(req_json.dump());
     return true;
 }
diff --git a/src/raft_server.cpp b/src/raft_server.cpp
index 560cd9ba..102df461 100644
--- a/src/raft_server.cpp
+++ b/src/raft_server.cpp
@@ -258,6 +258,20 @@ void ReplicationState::write(const std::shared_ptr<http_req>& request, const std
         return message_dispatcher->send_message(HttpServer::STREAM_RESPONSE_MESSAGE, req_res);
     }
 
+    route_path* rpath = nullptr;
+    bool route_found = server->get_route(request->route_hash, &rpath);
+
+    if(route_found && rpath->handler == patch_update_collection) {
+        if(get_alter_in_progress()) {
+            response->set_422("Another collection update operation is in progress.");
+            response->final = true;
+            auto req_res = new async_req_res_t(request, response, true);
+            return message_dispatcher->send_message(HttpServer::STREAM_RESPONSE_MESSAGE, req_res);
+        }
+
+        set_alter_in_progress(true);
+    }
+
     std::shared_lock lock(node_mutex);
 
     if(!node) {