Merge branch 'devel' of https://github.com/arangodb/arangodb into devel

arangod/Cluster/ClusterInfo.cpp

@@ -942,7 +942,7 @@ void ClusterInfo::loadCurrentCollections() {
       _currentCollections.swap(newCollections);
       _shardIds.swap(newShardIds);
       _currentCollectionsProt.version++;  // such that others notice our change
-      _currentCollectionsProt.isValid = true;  // will never be reset to false
+      _currentCollectionsProt.isValid = true;
     }
     return;
   }
@@ -2506,6 +2506,14 @@ std::vector<ServerID> ClusterInfo::getCurrentCoordinators() {
   return result;
 }
 
+//////////////////////////////////////////////////////////////////////////////
+/// @brief invalidate current
+//////////////////////////////////////////////////////////////////////////////
+void ClusterInfo::invalidateCurrent() {
+  WRITE_LOCKER(writeLocker, _currentCollectionsProt.lock);
+  _currentCollectionsProt.isValid = false;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief get information about current followers of a shard.
 ////////////////////////////////////////////////////////////////////////////////

arangod/Cluster/ClusterInfo.h

@@ -810,6 +810,12 @@ class ClusterInfo {
 
   std::vector<ServerID> getCurrentCoordinators();
 
+  //////////////////////////////////////////////////////////////////////////////
+  /// @brief invalidate current
+  //////////////////////////////////////////////////////////////////////////////
+  
+  void invalidateCurrent();
+
  private:
   //////////////////////////////////////////////////////////////////////////////
   /// @brief actually clears a list of planned databases

arangod/Cluster/HeartbeatThread.cpp

@@ -261,6 +261,8 @@ void HeartbeatThread::runCoordinator() {
 
   // last value of plan which we have noticed:
   uint64_t lastPlanVersionNoticed = 0;
+  // last value of current which we have noticed:
+  uint64_t lastCurrentVersionNoticed = 0;
 
   // value of Sync/Commands/my-id at startup
   uint64_t lastCommandIndex = getLastCommandIndex();
@@ -362,6 +364,28 @@ void HeartbeatThread::runCoordinator() {
       }
     }
 
+    result = _agency.getValues("Current/Version", false);
+    if (result.successful()) {
+      result.parse("", false);
+
+      std::map<std::string, AgencyCommResultEntry>::iterator it =
+          result._values.begin();
+
+      if (it != result._values.end()) {
+        // there is a plan version
+        uint64_t currentVersion =
+            arangodb::basics::VelocyPackHelper::stringUInt64(
+                it->second._vpack->slice());
+        
+        if (currentVersion > lastCurrentVersionNoticed) {
+          lastCurrentVersionNoticed = currentVersion;
+
+          ClusterInfo::instance()->invalidateCurrent();
+        }
+      }
+    }
+
+
     if (shouldSleep) {
       double remain = interval - (TRI_microtime() - start);
 

js/server/modules/@arangodb/cluster.js

@@ -557,12 +557,13 @@ function createLocalCollections (plannedCollections, planVersion, takeOverRespon
                             payload);
   };
   
-  // mop: just a function alias but this way one at least knows what it is supposed to do :S
   var takeOver = createCollectionAgency;
 
 
   var db = require("internal").db;
   db._useDatabase("_system");
+
+  var migrate = writeLocked => {
     var localDatabases = getLocalDatabases();
     var database;
     var i;
@@ -779,7 +780,6 @@ function createLocalCollections (plannedCollections, planVersion, takeOverRespon
                     }
 
                     if (takeOverResponsibility && !didWrite) {
-                    console.info("HMMMM WRITE");
                       writeLocked({ part: "Current" },
                           takeOver,
                           [ database, shard, collInfo, error ]);
@@ -803,6 +803,22 @@ function createLocalCollections (plannedCollections, planVersion, takeOverRespon
     }
   }
 
+  if (takeOverResponsibility) {
+    // mop: if this is a complete takeover we need a global lock because
+    // otherwise the coordinator might fetch results which are only partly
+    // migrated
+    var fakeLock = (lockInfo, cb, args) => {
+      if (!lockInfo || lockInfo.part != 'Current') {
+        throw new Error("Invalid lockInfo " + JSON.stringify(lockInfo));
+      }
+      return cb(...args);
+    }
+    writeLocked({ part: "Current" }, migrate, [fakeLock]);
+  } else {
+    migrate(writeLocked);
+  }
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief drop collections if they exist locally but not in the plan
 ////////////////////////////////////////////////////////////////////////////////

scripts/startLocalCluster.sh

@@ -57,7 +57,6 @@ start() {
                 --cluster.my-role $ROLE \
                 --log.file cluster/$PORT.log \
                 --log.requests-file cluster/$PORT.req \
-                --log.level TRACE \
                 --server.disable-statistics true \
                 --server.foxx-queues false \
                 --javascript.startup-directory ./js \