Rollback request if storing log fails

contrib/NuRaft

@@ -1 +1 @@
-Subproject commit e1dc47c1cfd529801a8c94a396a3921a71ae3ccf
+Subproject commit bef4644837ba7ef9ced2ccf5c750a936c58d424d

src/Coordination/KeeperServer.cpp

@@ -573,6 +573,19 @@ nuraft::cb_func::ReturnCode KeeperServer::callbackFunc(nuraft::cb_func::Type typ
                 entry = nuraft::cs_new<nuraft::log_entry>(entry->get_term(), getZooKeeperLogEntry(request_for_session), entry->get_val_type());
                 break;
             }
+            case nuraft::cb_func::AppendLogFailed:
+            {
+                // we are relying on the fact that request are being processed under a mutex
+                // and not a RW lock
+                auto & entry = *static_cast<LogEntryPtr *>(param->ctx);
+
+                assert(entry->get_val_type() == nuraft::app_log);
+
+                auto & entry_buf = entry->get_buf();
+                auto request_for_session = state_machine->parseRequest(entry_buf);
+                state_machine->rollbackRequest(request_for_session);
+                break;
+            }
             default:
                 break;
         }

src/Coordination/KeeperStateMachine.cpp

@@ -311,6 +311,11 @@ void KeeperStateMachine::rollback(uint64_t log_idx, nuraft::buffer & data)
     if (!request_for_session.zxid)
         request_for_session.zxid = log_idx;
 
+    rollbackRequest(request_for_session);
+}
+
+void KeeperStateMachine::rollbackRequest(const KeeperStorage::RequestForSession & request_for_session)
+{
     if (request_for_session.request->getOpNum() == Coordination::OpNum::SessionID)
         return;
 

src/Coordination/KeeperStateMachine.h

@@ -44,6 +44,8 @@ public:
 
     void rollback(uint64_t log_idx, nuraft::buffer & data) override;
 
+    void rollbackRequest(const KeeperStorage::RequestForSession & request_for_session);
+
     uint64_t last_commit_index() override { return last_committed_idx; }
 
     /// Apply preliminarily saved (save_logical_snp_obj) snapshot to our state.

src/Coordination/KeeperStorage.cpp

@@ -6,7 +6,7 @@
 #include <boost/algorithm/string.hpp>
 #include <Poco/Base64Encoder.h>
 #include <Poco/SHA1Engine.h>
-#include "Common/ZooKeeper/ZooKeeperCommon.h"
+#include <Common/ZooKeeper/ZooKeeperCommon.h>
 #include <Common/SipHash.h>
 #include <Common/ZooKeeper/ZooKeeperConstants.h>
 #include <Common/StringUtils/StringUtils.h>
@@ -14,6 +14,7 @@
 #include <Common/hex.h>
 #include <Common/logger_useful.h>
 #include <Common/setThreadName.h>
+#include <Common/MemoryTrackerBlockerInThread.h>
 #include <Coordination/pathUtils.h>
 #include <Coordination/KeeperConstants.h>
 #include <sstream>
@@ -2119,8 +2120,19 @@ void KeeperStorage::rollbackRequest(int64_t rollback_zxid)
         throw Exception(
             ErrorCodes::LOGICAL_ERROR, "Trying to rollback invalid ZXID ({}). It should be the last preprocessed.", rollback_zxid);
 
-    uncommitted_transactions.pop_back();
-    uncommitted_state.rollback(rollback_zxid);
+    // if an exception occurs during rollback, the best option is to terminate because we can end up in an incosistent state
+    // we block memory tracking so we can avoid terminating if we're rollbacking because of memory limit
+    MemoryTrackerBlockerInThread temporarily_ignore_any_memory_limits(VariableContext::Global);
+    try
+    {
+        uncommitted_transactions.pop_back();
+        uncommitted_state.rollback(rollback_zxid);
+    }
+    catch (...)
+    {
+        LOG_FATAL(&Poco::Logger::get("KeeperStorage"), "Failed to rollback log. Terminating to avoid incosistencies");
+        std::terminate();
+    }
 }
 
 KeeperStorage::Digest KeeperStorage::getNodesDigest(bool committed) const