First stab at following term id.

arangodb · neunhoef · Jul 7, 2021 · Jun 25, 2021 · Jun 25, 2021 · Jun 25, 2021
commit e8e551b36b614bacf859b8c6fee7dd9fc621fa27
diff --git a/arangod/Cluster/FollowerInfo.cpp b/arangod/Cluster/FollowerInfo.cpp
@@ -33,6 +33,7 @@
 #include "Logger/LogMacros.h"
 #include "Logger/Logger.h"
 #include "Logger/LoggerStream.h"
+#include "Random/RandomGenerator.h"
 #include "VocBase/LogicalCollection.h"
 
 using namespace arangodb;
@@ -542,3 +543,25 @@ VPackBuilder FollowerInfo::newShardEntry(VPackSlice oldValue) const {
   }
   return newValue;
 }
+
+uint64_t FollowerInfo::newFollowingTermId(ServerID const& s) noexcept {
+  uint64_t i = RandomGenerator::interval(UINT64_MAX);
+  try {
+    _followingTermId[s] = i;
+  } catch(std::bad_alloc const& exc) {
+    i = 0;   // I assume here that I do not get bad_alloc if the key is
+             // already in the map, since it then only has to overwrite
+             // an integer, if the key is not in the map, we default to 0.
+  }
+  return i;
+}
+
+uint64_t FollowerInfo::getFollowingTermId(ServerID const& s) noexcept {
+  // Note that we assume that find() does not throw!
+  auto it = _followingTermId.find(s);
+  if (it == _followingTermId.end()) {
+    return 0;
+  }
+  return it->second;
+}
+
diff --git a/arangod/Cluster/FollowerInfo.h b/arangod/Cluster/FollowerInfo.h
@@ -54,6 +54,20 @@ class FollowerInfo {
   // soon as we can guarantee at least so many followers locally.
   std::shared_ptr<std::vector<ServerID>> _failoverCandidates;
 
+  // The following map holds a random number for each follower, this
+  // random number is sent given to the follower when it gets in sync
+  // (actually, when it acquires the hard lock to get in sync), and is
+  // then subsequently sent alongside every synchronous replication
+  // request. If the number does not match, the follower will refuse the
+  // replication request. This is to ensure that replication requests cannot
+  // be delayed into the "next" leader/follower relationship.
+  // And here is the proof that this works: The exclusive lock divides the
+  // write operations between "before" and "after". The id is changed
+  // when the exclusive lock is established. Therefore, it is OK for the
+  // replication requests to send along the "current" id, directly
+  // from this map.
+  std::unordered_map<ServerID, uint64_t> _followingTermId;
+
   // The agencyMutex is used to synchronise access to the agency.
   // the _dataLock is used to sync the access to local data.
   // The _canWriteLock is used to protect flag if we do have enough followers
@@ -135,6 +149,28 @@ class FollowerInfo {
 
   Result remove(ServerID const& s);
 
+  //////////////////////////////////////////////////////////////////////////////
+  /// @brief for each run of the "get-in-sync" protocol we generate a
+  /// random number to identify this "following term". This is created
+  /// when the follower fetches the exclusive lock to finally get in sync
+  /// and is stored in _followingTermId, so that it can be forwarded with
+  /// each synchronous replication request. The follower can then decline
+  /// the replication in case it is not "in the same term".
+  //////////////////////////////////////////////////////////////////////////////
+
+  uint64_t newFollowingTermId(ServerID const& s) noexcept;
+
+  //////////////////////////////////////////////////////////////////////////////
+  /// @brief for each run of the "get-in-sync" protocol we generate a
+  /// random number to identify this "following term". This is created
+  /// when the follower fetches the exclusive lock to finally get in sync
+  /// and is stored in _followingTermId, so that it can be forwarded with
+  /// each synchronous replication request. The follower can then decline
+  /// the replication in case it is not "in the same term".
+  //////////////////////////////////////////////////////////////////////////////
+
+  uint64_t getFollowingTermId(ServerID const& s) noexcept;
+
   //////////////////////////////////////////////////////////////////////////////
   /// @brief clear follower list, no changes in agency necesary
   //////////////////////////////////////////////////////////////////////////////

diff --git a/arangod/Cluster/SynchronizeShard.cpp b/arangod/Cluster/SynchronizeShard.cpp
@@ -92,7 +92,8 @@ using namespace std::chrono;
 SynchronizeShard::SynchronizeShard(MaintenanceFeature& feature, ActionDescription const& desc)
   : ActionBase(feature, desc),
     ShardDefinition(desc.get(DATABASE), desc.get(SHARD)),
-    _leaderInfo(arangodb::replutils::LeaderInfo::createEmpty()) {
+    _leaderInfo(arangodb::replutils::LeaderInfo::createEmpty()),
+    _followingTermId(0) {
   std::stringstream error;
 
   if (!desc.has(COLLECTION)) {
@@ -416,7 +417,7 @@ arangodb::Result SynchronizeShard::getReadLock(
   // nullptr only happens during controlled shutdown
   if (pool == nullptr) {
     return arangodb::Result(TRI_ERROR_SHUTTING_DOWN,
-                            "cancelReadLockOnLeader: Shutting down");
+                            "getReadLock: Shutting down");
   }
 
   VPackBuilder body;
@@ -445,6 +446,16 @@ arangodb::Result SynchronizeShard::getReadLock(
 
   if (res.ok()) {
     // Habemus clausum, we have a lock
+    if (!soft) {
+      // Now store the random followingTermId:
+      VPackSlice body = response.response().slice();
+      if (body.isObject()) {
+        VPackSlice followingTermIdSlice = body.get(StaticStrings::FollowingTermId);
+        if (followingTermIdSlice.isNumber()) {
+          _followingTermId = followingTermIdSlice.getNumber<uint64_t>();
+        }
+      }
+    }
     return arangodb::Result();
   }
 
@@ -913,10 +924,6 @@ bool SynchronizeShard::first() {
         return false;
       }
 
-      // This is necessary to accept replications from the leader which can
-      // happen as soon as we are in sync.
-      collection->followers()->setTheLeader(leader);
-
       startTime = system_clock::now();
 
       VPackBuilder config;
@@ -1169,6 +1176,13 @@ Result SynchronizeShard::catchupWithExclusiveLock(
       }
     });
 
+  // Now we have got a unique id for this following term and have stored it
+  // in _followingTermId, so we can use it to set the leader:
+
+  // This is necessary to accept replications from the leader which can
+  // happen as soon as we are in sync.
+  collection.followers()->setTheLeader(leader + "_" + basics::StringUtils::itoa(_followingTermId));
+
      LOG_TOPIC("d76cb", DEBUG, Logger::MAINTENANCE) << "lockJobId: " << lockJobId;
 
   builder.clear();

diff --git a/arangod/Cluster/SynchronizeShard.h b/arangod/Cluster/SynchronizeShard.h
@@ -91,6 +91,7 @@ class SynchronizeShard : public ActionBase, public ShardDefinition {
 
   /// @brief information about the leader, reused across multiple replication steps
   arangodb::replutils::LeaderInfo _leaderInfo;
+  uint64_t _followingTermId;
 };
 
 }  // namespace maintenance

diff --git a/arangod/RestHandler/RestReplicationHandler.cpp b/arangod/RestHandler/RestReplicationHandler.cpp
@@ -2708,6 +2708,10 @@ void RestReplicationHandler::handleCommandHoldReadLockCollection() {
   {
     VPackObjectBuilder bb(&b);
     b.add(StaticStrings::Error, VPackValue(false));
+    if (!serverId.empty()) {
+      b.add(StaticStrings::FollowingTermId,
+            VPackValue(col->followers()->newFollowingTermId(serverId)));
+    }
   }
 
   LOG_TOPIC("61a9d", DEBUG, Logger::REPLICATION)

diff --git a/arangod/Transaction/Methods.cpp b/arangod/Transaction/Methods.cpp
@@ -1221,7 +1221,8 @@ Future<OperationResult> transaction::Methods::insertLocal(std::string const& cna
 
     // Now replicate the good operations on all followers:
     return replicateOperations(collection.get(), followers, options, value,
-                               TRI_VOC_DOCUMENT_OPERATION_INSERT, resDocs)
+                               TRI_VOC_DOCUMENT_OPERATION_INSERT, resDocs,
+                               *collection->followers())
     .thenValue([options, errs = std::move(errorCounter), resDocs](Result res) mutable {
       if (!res.ok()) {
         return OperationResult{std::move(res), options};
@@ -1493,7 +1494,7 @@ Future<OperationResult> transaction::Methods::modifyLocal(std::string const& col
 
     // Now replicate the good operations on all followers:
     return replicateOperations(collection.get(), followers, options, newValue,
-                               operation, resDocs)
+                               operation, resDocs, *collection->followers())
     .thenValue([options, errs = std::move(errorCounter), resDocs](Result&& res) mutable {
       if (!res.ok()) {
         return OperationResult{std::move(res), options};
@@ -1704,7 +1705,8 @@ Future<OperationResult> transaction::Methods::removeLocal(std::string const& col
 
     // Now replicate the good operations on all followers:
     return replicateOperations(collection.get(), followers, options, value,
-                               TRI_VOC_DOCUMENT_OPERATION_REMOVE, resDocs)
+                               TRI_VOC_DOCUMENT_OPERATION_REMOVE, resDocs,
+                               *collection->followers())
     .thenValue([options, errs = std::move(errorCounter), resDocs](Result res) mutable {
       if (!res.ok()) {
         return OperationResult{std::move(res), options};
@@ -1892,10 +1894,15 @@ Future<OperationResult> transaction::Methods::truncateLocal(std::string const& c
       network::RequestOptions reqOpts;
       reqOpts.database = vocbase().name();
       reqOpts.timeout = network::Timeout(600);
-      reqOpts.param(StaticStrings::IsSynchronousReplicationString, ServerState::instance()->getId());
       reqOpts.param(StaticStrings::Compact, (options.truncateCompact ? "true" : "false"));
 
       for (auto const& f : *followers) {
+        reqOpts.param(StaticStrings::IsSynchronousReplicationString,
+            ServerState::instance()->getId() + "_" +
+            basics::StringUtils::itoa(
+              collection->followers()->getFollowingTermId(f)));
+        // reqOpts is copied deep in sendRequestRetry, so we are OK to
+        // change it in the loop!
         network::Headers headers;
         ClusterTrxMethods::addTransactionHeader(*this, f, headers);
         auto future = network::sendRequestRetry(pool, "server:" + f, fuerte::RestVerb::Put,
@@ -2311,7 +2318,8 @@ Future<Result> Methods::replicateOperations(
     std::shared_ptr<const std::vector<ServerID>> const& followerList,
     OperationOptions const& options, VPackSlice const value,
     TRI_voc_document_operation_e const operation,
-    std::shared_ptr<VPackBuffer<uint8_t>> const& ops) {
+    std::shared_ptr<VPackBuffer<uint8_t>> const& ops,
+    FollowerInfo& followerInfo) {
   TRI_ASSERT(followerList != nullptr);
 
   if (followerList->empty()) {
@@ -2323,8 +2331,6 @@ Future<Result> Methods::replicateOperations(
   network::RequestOptions reqOpts;
   reqOpts.database = vocbase().name();
   reqOpts.param(StaticStrings::IsRestoreString, "true");
-  reqOpts.param(StaticStrings::IsSynchronousReplicationString, ServerState::instance()->getId());
-
   std::string url = "/_api/document/";
   url.append(arangodb::basics::StringUtils::urlEncode(collection->name()));
   if (operation != TRI_VOC_DOCUMENT_OPERATION_INSERT && !value.isArray()) {
@@ -2417,6 +2423,12 @@ Future<Result> Methods::replicateOperations(
 
   auto* pool = vocbase().server().getFeature<NetworkFeature>().pool();
   for (auto const& f : *followerList) {
+    reqOpts.param(StaticStrings::IsSynchronousReplicationString,
+        ServerState::instance()->getId() + "_" +
+        basics::StringUtils::itoa(
+          collection->followers()->getFollowingTermId(f)));
+    // reqOpts is copied deep in sendRequestRetry, so we are OK to
+    // change it in the loop!
     network::Headers headers;
     ClusterTrxMethods::addTransactionHeader(*this, f, headers);
     futures.emplace_back(network::sendRequestRetry(pool, "server:" + f, requestType,

diff --git a/arangod/Transaction/Methods.h b/arangod/Transaction/Methods.h
@@ -27,6 +27,7 @@
 #include "Basics/Common.h"
 #include "Basics/Exceptions.h"
 #include "Basics/Result.h"
+#include "Cluster/FollowerInfo.h"
 #include "Futures/Future.h"
 #include "Indexes/IndexIterator.h"
 #include "Rest/CommonDefines.h"
@@ -492,7 +493,8 @@ class Methods {
       LogicalCollection* collection,
       std::shared_ptr<const std::vector<std::string>> const& followers,
       OperationOptions const& options, VPackSlice value, TRI_voc_document_operation_e operation,
-      std::shared_ptr<velocypack::Buffer<uint8_t>> const& ops);
+      std::shared_ptr<velocypack::Buffer<uint8_t>> const& ops,
+      FollowerInfo& followerInfo);
 
  private:
   /// @brief transaction hints

diff --git a/lib/Basics/StaticStrings.cpp b/lib/Basics/StaticStrings.cpp
@@ -336,6 +336,7 @@ std::string const StaticStrings::RevisionTreeInitialRangeMin("initialRangeMin");
 std::string const StaticStrings::RevisionTreeRanges("ranges");
 std::string const StaticStrings::RevisionTreeResume("resume");
 std::string const StaticStrings::RevisionTreeVersion("version");
+std::string const StaticStrings::FollowingTermId("followingTermId");
 
 // Generic attribute names
 std::string const StaticStrings::AttrCoordinator("coordinator");

diff --git a/lib/Basics/StaticStrings.h b/lib/Basics/StaticStrings.h
@@ -312,6 +312,7 @@ class StaticStrings {
   static std::string const RevisionTreeRanges;
   static std::string const RevisionTreeResume;
   static std::string const RevisionTreeVersion;
+  static std::string const FollowingTermId;
 
   // generic attribute names
   static std::string const AttrCoordinator;