diff --git a/arangod/Agency/AgentConfiguration.cpp b/arangod/Agency/AgentConfiguration.cpp
index 071b5c3d5f..0e0375c42a 100644
--- a/arangod/Agency/AgentConfiguration.cpp
+++ b/arangod/Agency/AgentConfiguration.cpp
@@ -74,7 +74,7 @@ config_t::config_t(
     _lock() {}
 
 config_t::config_t(config_t const& other) { 
-  READ_LOCKER(readLocker, other._lock);
+  // will call operator=, which will ensure proper locking
   *this = other; 
 }
 
diff --git a/arangod/Agency/RestAgencyHandler.cpp b/arangod/Agency/RestAgencyHandler.cpp
index e8ca07b274..4ff38cc103 100644
--- a/arangod/Agency/RestAgencyHandler.cpp
+++ b/arangod/Agency/RestAgencyHandler.cpp
@@ -88,7 +88,7 @@ void RestAgencyHandler::redirectRequest(std::string const& leaderId) {
     _response->setResponseCode(rest::ResponseCode::TEMPORARY_REDIRECT);
     _response->setHeaderNC(StaticStrings::Location, url);
     LOG_TOPIC(DEBUG, Logger::AGENCY) << "Sending 307 redirect to " << url;
-  } catch (std::exception const& e) {
+  } catch (std::exception const&) {
     reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
   }
 }
diff --git a/arangod/Aql/Ast.cpp b/arangod/Aql/Ast.cpp
index bcd635cf1d..9c4d57b718 100644
--- a/arangod/Aql/Ast.cpp
+++ b/arangod/Aql/Ast.cpp
@@ -122,7 +122,7 @@ std::shared_ptr<VPackBuilder> Ast::toVelocyPack(bool verbose) const {
   return builder;
 }
 
-/// @brief destroy the AST
+/// @brief add an operation to the AST
 void Ast::addOperation(AstNode* node) {
   TRI_ASSERT(_root != nullptr);
 
@@ -840,6 +840,7 @@ AstNode* Ast::createNodeValueInt(int64_t value) {
   AstNode* node = createNode(NODE_TYPE_VALUE);
   node->setValueType(VALUE_TYPE_INT);
   node->setIntValue(value);
+  node->setFlag(DETERMINED_CONSTANT, VALUE_CONSTANT);
 
   return node;
 }
@@ -849,6 +850,7 @@ AstNode* Ast::createNodeValueDouble(double value) {
   AstNode* node = createNode(NODE_TYPE_VALUE);
   node->setValueType(VALUE_TYPE_DOUBLE);
   node->setDoubleValue(value);
+  node->setFlag(DETERMINED_CONSTANT, VALUE_CONSTANT);
 
   return node;
 }
@@ -869,6 +871,7 @@ AstNode* Ast::createNodeValueString(char const* value, size_t length) {
   AstNode* node = createNode(NODE_TYPE_VALUE);
   node->setValueType(VALUE_TYPE_STRING);
   node->setStringValue(value, length);
+  node->setFlag(DETERMINED_CONSTANT, VALUE_CONSTANT);
 
   return node;
 }
@@ -1247,13 +1250,13 @@ AstNode* Ast::createNodeShortestPath(
 }
 
 /// @brief create an AST function call node
-AstNode* Ast::createNodeFunctionCall(char const* functionName,
+AstNode* Ast::createNodeFunctionCall(char const* functionName, size_t length,
                                      AstNode const* arguments) {
   if (functionName == nullptr) {
     THROW_ARANGO_EXCEPTION(TRI_ERROR_OUT_OF_MEMORY);
   }
 
-  auto normalized = normalizeFunctionName(functionName);
+  auto normalized = normalizeFunctionName(functionName, length);
 
   AstNode* node;
 
@@ -2946,7 +2949,7 @@ AstNode* Ast::optimizeFunctionCall(AstNode* node) {
         auto countArgs = createNodeArray();
         countArgs->addMember(createNodeValueString(arg->getStringValue(),
                                                    arg->getStringLength()));
-        return createNodeFunctionCall("COLLECTION_COUNT", countArgs);
+        return createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("COLLECTION_COUNT"), countArgs);
       }
     }
   } else if (func->name == "IS_NULL") {
@@ -3180,6 +3183,11 @@ AstNode* Ast::nodeFromVPack(VPackSlice const& slice, bool copyStringValues) {
   }
 
   if (slice.isNumber()) {
+    if (slice.isSmallInt() || slice.isInt()) {
+      // integer value
+      return createNodeValueInt(slice.getInt());
+    }
+    // floating point value
     return createNodeValueDouble(slice.getNumber<double>());
   }
 
@@ -3394,10 +3402,10 @@ void Ast::traverseReadOnly(AstNode const* node,
 }
 
 /// @brief normalize a function name
-std::pair<std::string, bool> Ast::normalizeFunctionName(char const* name) {
+std::pair<std::string, bool> Ast::normalizeFunctionName(char const* name, size_t length) {
   TRI_ASSERT(name != nullptr);
 
-  std::string functionName(name);
+  std::string functionName(name, length);
   // convert name to upper case
   std::transform(functionName.begin(), functionName.end(), functionName.begin(), ::toupper);
 
diff --git a/arangod/Aql/Ast.h b/arangod/Aql/Ast.h
index 5c2db24bea..e3bfa1a733 100644
--- a/arangod/Aql/Ast.h
+++ b/arangod/Aql/Ast.h
@@ -344,7 +344,11 @@ class Ast {
                                   AstNode const*, AstNode const*);
 
   /// @brief create an AST function call node
-  AstNode* createNodeFunctionCall(char const*, AstNode const*);
+  AstNode* createNodeFunctionCall(char const* functionName, AstNode const* arguments) {
+    return createNodeFunctionCall(functionName, strlen(functionName), arguments);
+  }
+
+  AstNode* createNodeFunctionCall(char const* functionName, size_t length, AstNode const* arguments);
 
   /// @brief create an AST range node
   AstNode* createNodeRange(AstNode const*, AstNode const*);
@@ -512,9 +516,9 @@ public:
   static void traverseReadOnly(AstNode const*,
                                std::function<void(AstNode const*, void*)>,
                                void*);
-private:
+ private:
   /// @brief normalize a function name
-  std::pair<std::string, bool> normalizeFunctionName(char const*);
+  std::pair<std::string, bool> normalizeFunctionName(char const* functionName, size_t length);
 
   /// @brief create a node of the specified type
   AstNode* createNode(AstNodeType);
diff --git a/arangod/Aql/Condition.cpp b/arangod/Aql/Condition.cpp
index 31749ef4c0..438a9e47d6 100644
--- a/arangod/Aql/Condition.cpp
+++ b/arangod/Aql/Condition.cpp
@@ -397,7 +397,17 @@ std::pair<bool, bool> Condition::findIndexes(
 
   transaction::Methods* trx = _ast->query()->trx();
 
-  size_t const itemsInIndex = node->collection()->count(trx);
+  size_t itemsInIndex;
+  if (!collectionName.empty() && collectionName[0] == '_' &&
+      collectionName.substr(0, 11) == "_statistics") {
+    // use hard-coded number of items in index, because we are dealing with
+    // the statistics collection here. this saves a roundtrip to the DB servers
+    // for statistics queries that do not need a fully accurate collection count
+    itemsInIndex = 1024;
+  } else {
+    // actually count number of items in index
+    itemsInIndex = node->collection()->count(trx);
+  }
   if (_root == nullptr) {
     size_t dummy;
     return trx->getIndexForSortCondition(collectionName, sortCondition,
diff --git a/arangod/Aql/IndexBlock.cpp b/arangod/Aql/IndexBlock.cpp
index 848fa90128..973c263a52 100644
--- a/arangod/Aql/IndexBlock.cpp
+++ b/arangod/Aql/IndexBlock.cpp
@@ -107,10 +107,10 @@ arangodb::aql::AstNode* IndexBlock::makeUnique(
     if (isSparse) {
       // the index is sorted. we need to use SORTED_UNIQUE to get the
       // result back in index order
-      return ast->createNodeFunctionCall("SORTED_UNIQUE", array);
+      return ast->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("SORTED_UNIQUE"), array);
     }
     // a regular UNIQUE will do
-    return ast->createNodeFunctionCall("UNIQUE", array);
+    return ast->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("UNIQUE"), array);
   }
 
   // presumably an array with no or a single member
diff --git a/arangod/Aql/OptimizerRules.cpp b/arangod/Aql/OptimizerRules.cpp
index f3d41f1f55..06a78aa227 100644
--- a/arangod/Aql/OptimizerRules.cpp
+++ b/arangod/Aql/OptimizerRules.cpp
@@ -226,7 +226,7 @@ void arangodb::aql::sortInValuesRule(Optimizer* opt,
 
     auto args = ast->createNodeArray();
     args->addMember(originalArg);
-    auto sorted = ast->createNodeFunctionCall("SORTED_UNIQUE", args);
+    auto sorted = ast->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("SORTED_UNIQUE"), args);
 
     auto outVar = ast->variables()->createTemporaryVariable();
     ExecutionNode* calculationNode = nullptr;
@@ -4983,10 +4983,10 @@ std::unique_ptr<Condition> buildGeoCondition(ExecutionPlan* plan,
     args->addMember(info.range);
     auto lessValue = ast->createNodeValueBool(info.lessgreaterequal);
     args->addMember(lessValue);
-    cond = ast->createNodeFunctionCall("WITHIN", args);
+    cond = ast->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("WITHIN"), args);
   } else {
     // NEAR
-    cond = ast->createNodeFunctionCall("NEAR", args);
+    cond = ast->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("NEAR"), args);
   }
 
   TRI_ASSERT(cond != nullptr);
diff --git a/arangod/Aql/QueryResources.cpp b/arangod/Aql/QueryResources.cpp
index 95dcf4a545..9b867f5246 100644
--- a/arangod/Aql/QueryResources.cpp
+++ b/arangod/Aql/QueryResources.cpp
@@ -56,8 +56,8 @@ QueryResources::~QueryResources() {
   _resourceMonitor->decreaseMemoryUsage(_nodes.size() * sizeof(AstNode) + _nodes.capacity() * sizeof(AstNode*));
 }
 
-// TODO: FIXME
 void QueryResources::steal() {
+  // we are not responsible for freeing any data, so we delete our inventory
   _strings.clear();
   _nodes.clear();
 }
@@ -103,7 +103,7 @@ char* QueryResources::registerString(char const* p, size_t length) {
     return const_cast<char*>(EmptyString);
   }
 
-  if (length < ShortStringStorage::MaxStringLength) {
+  if (length < ShortStringStorage::maxStringLength()) {
     return _shortStringStorage.registerString(p, length);
   }
 
@@ -148,10 +148,15 @@ char* QueryResources::registerLongString(char* copy, size_t length) {
     TRI_ASSERT(capacity >= _strings.capacity());
     
     // reserve space
-    _resourceMonitor->increaseMemoryUsage((capacity - _strings.size()) * sizeof(char*));
-    _strings.reserve(capacity);
+    _resourceMonitor->increaseMemoryUsage(((capacity - _strings.size()) * sizeof(char*)) + length);
+    try {
+      _strings.reserve(capacity);
+    } catch (...) {
+      // revert change in memory increase
+      _resourceMonitor->decreaseMemoryUsage(((capacity - _strings.size()) * sizeof(char*)) + length);
+      throw;
+    }
     
-    _resourceMonitor->increaseMemoryUsage(length);
     // will not fail 
     _strings.emplace_back(copy);
     _stringsLength += length;
diff --git a/arangod/Aql/QueryResources.h b/arangod/Aql/QueryResources.h
index 11255ab93f..eb08e3dacd 100644
--- a/arangod/Aql/QueryResources.h
+++ b/arangod/Aql/QueryResources.h
@@ -49,17 +49,17 @@ class QueryResources {
   
   /// @brief register a string
   /// the string is freed when the query is destroyed
-  char* registerString(char const*, size_t);
+  char* registerString(char const* p, size_t length);
 
   /// @brief register a string
   /// the string is freed when the query is destroyed
   char* registerString(std::string const& p) {
-    return registerString(p.c_str(), p.length());
+    return registerString(p.data(), p.size());
   }
 
   /// @brief register a potentially UTF-8-escaped string
   /// the string is freed when the query is destroyed
-  char* registerEscapedString(char const*, size_t, size_t&);
+  char* registerEscapedString(char const* p, size_t length, size_t& outLength);
 
  private:
   char* registerLongString(char* copy, size_t length);
diff --git a/arangod/Aql/ShortStringStorage.cpp b/arangod/Aql/ShortStringStorage.cpp
index 5fbddd245d..d894173221 100644
--- a/arangod/Aql/ShortStringStorage.cpp
+++ b/arangod/Aql/ShortStringStorage.cpp
@@ -27,9 +27,6 @@
 
 using namespace arangodb::aql;
 
-/// @brief maximum length of a "short" string
-size_t const ShortStringStorage::MaxStringLength = 127;
-
 /// @brief create a short string storage instance
 ShortStringStorage::ShortStringStorage(ResourceMonitor* resourceMonitor, size_t blockSize)
     : _resourceMonitor(resourceMonitor), _blocks(), _blockSize(blockSize), _current(nullptr), _end(nullptr) {
@@ -46,7 +43,7 @@ ShortStringStorage::~ShortStringStorage() {
 
 /// @brief register a short string
 char* ShortStringStorage::registerString(char const* p, size_t length) {
-  TRI_ASSERT(length <= MaxStringLength);
+  TRI_ASSERT(length <= maxStringLength());
 
   if (_current == nullptr || (_current + length + 1 > _end)) {
     allocateBlock();
diff --git a/arangod/Aql/ShortStringStorage.h b/arangod/Aql/ShortStringStorage.h
index 466e67b94b..fc22601d13 100644
--- a/arangod/Aql/ShortStringStorage.h
+++ b/arangod/Aql/ShortStringStorage.h
@@ -48,7 +48,7 @@ class ShortStringStorage {
 
  public:
   /// @brief maximum length of strings in short string storage
-  static size_t const MaxStringLength;
+  static constexpr size_t maxStringLength() { return 127; }
 
  private:
   ResourceMonitor* _resourceMonitor;
diff --git a/arangod/Aql/grammar.cpp b/arangod/Aql/grammar.cpp
index 5661e8a878..2204f51a35 100644
--- a/arangod/Aql/grammar.cpp
+++ b/arangod/Aql/grammar.cpp
@@ -3244,7 +3244,7 @@ yyreduce:
 #line 1111 "Aql/grammar.y" /* yacc.c:1646  */
     {
       auto list = static_cast<AstNode const*>(parser->popStack());
-      (yyval.node) = parser->ast()->createNodeFunctionCall("LIKE", list);
+      (yyval.node) = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("LIKE"), list);
     }
 #line 3250 "Aql/grammar.cpp" /* yacc.c:1646  */
     break;
@@ -3399,7 +3399,7 @@ yyreduce:
       AstNode* arguments = parser->ast()->createNodeArray(2);
       arguments->addMember((yyvsp[-2].node));
       arguments->addMember((yyvsp[0].node));
-      (yyval.node) = parser->ast()->createNodeFunctionCall("LIKE", arguments);
+      (yyval.node) = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("LIKE"), arguments);
     }
 #line 3405 "Aql/grammar.cpp" /* yacc.c:1646  */
     break;
@@ -3410,7 +3410,7 @@ yyreduce:
       AstNode* arguments = parser->ast()->createNodeArray(2);
       arguments->addMember((yyvsp[-2].node));
       arguments->addMember((yyvsp[0].node));
-      (yyval.node) = parser->ast()->createNodeFunctionCall("REGEX_TEST", arguments);
+      (yyval.node) = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("REGEX_TEST"), arguments);
     }
 #line 3416 "Aql/grammar.cpp" /* yacc.c:1646  */
     break;
@@ -3421,7 +3421,7 @@ yyreduce:
       AstNode* arguments = parser->ast()->createNodeArray(2);
       arguments->addMember((yyvsp[-2].node));
       arguments->addMember((yyvsp[0].node));
-      AstNode* node = parser->ast()->createNodeFunctionCall("REGEX_TEST", arguments);
+      AstNode* node = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("REGEX_TEST"), arguments);
       (yyval.node) = parser->ast()->createNodeUnaryOperator(NODE_TYPE_OPERATOR_UNARY_NOT, node);
     }
 #line 3428 "Aql/grammar.cpp" /* yacc.c:1646  */
diff --git a/arangod/Aql/grammar.y b/arangod/Aql/grammar.y
index d1a16f0acc..6d07bee9cf 100644
--- a/arangod/Aql/grammar.y
+++ b/arangod/Aql/grammar.y
@@ -1115,7 +1115,7 @@ function_call:
       parser->pushStack(node);
     } optional_function_call_arguments T_CLOSE %prec FUNCCALL {
       auto list = static_cast<AstNode const*>(parser->popStack());
-      $$ = parser->ast()->createNodeFunctionCall("LIKE", list);
+      $$ = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("LIKE"), list);
     } 
   ;
 
@@ -1181,19 +1181,19 @@ operator_binary:
       AstNode* arguments = parser->ast()->createNodeArray(2);
       arguments->addMember($1);
       arguments->addMember($3);
-      $$ = parser->ast()->createNodeFunctionCall("LIKE", arguments);
+      $$ = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("LIKE"), arguments);
     }
   | expression T_REGEX_MATCH expression {
       AstNode* arguments = parser->ast()->createNodeArray(2);
       arguments->addMember($1);
       arguments->addMember($3);
-      $$ = parser->ast()->createNodeFunctionCall("REGEX_TEST", arguments);
+      $$ = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("REGEX_TEST"), arguments);
     }
   | expression T_REGEX_NON_MATCH expression {
       AstNode* arguments = parser->ast()->createNodeArray(2);
       arguments->addMember($1);
       arguments->addMember($3);
-      AstNode* node = parser->ast()->createNodeFunctionCall("REGEX_TEST", arguments);
+      AstNode* node = parser->ast()->createNodeFunctionCall(TRI_CHAR_LENGTH_PAIR("REGEX_TEST"), arguments);
       $$ = parser->ast()->createNodeUnaryOperator(NODE_TYPE_OPERATOR_UNARY_NOT, node);
     }
   | expression quantifier T_EQ expression {
diff --git a/arangod/Cluster/ClusterComm.h b/arangod/Cluster/ClusterComm.h
index e9cbd8a493..728fc7ca1e 100644
--- a/arangod/Cluster/ClusterComm.h
+++ b/arangod/Cluster/ClusterComm.h
@@ -539,7 +539,7 @@ class ClusterComm {
  protected:  // protected members are for unit test purposes
 
   /// @brief Constructor for test cases.
-  ClusterComm(bool);
+  explicit ClusterComm(bool);
 
   // code below this point used to be "private".  now "protected" to
   //  enable unit test wrapper class
diff --git a/arangod/Cluster/ServerState.cpp b/arangod/Cluster/ServerState.cpp
index 67ef3531c7..ba25c22de7 100644
--- a/arangod/Cluster/ServerState.cpp
+++ b/arangod/Cluster/ServerState.cpp
@@ -826,7 +826,7 @@ bool ServerState::storeRole(RoleEnum role) {
       AgencyComm comm; // should not throw anything
       AgencyCommResult res = comm.sendTransactionWithFailover(*trx.get(), 1.0);
       if (!res.successful()) {
-          return false;
+        return false;
       }
     } catch (...) {
       LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << __FUNCTION__
diff --git a/arangod/Cluster/TraverserEngine.cpp b/arangod/Cluster/TraverserEngine.cpp
index 086b08ea64..0c95711e85 100644
--- a/arangod/Cluster/TraverserEngine.cpp
+++ b/arangod/Cluster/TraverserEngine.cpp
@@ -166,13 +166,16 @@ bool BaseEngine::lockCollection(std::string const& shard) {
     return false;
   }
   _trx->pinData(cid);  // will throw when it fails
-  Result res = _trx->lock(cid, AccessMode::Type::READ);
-  if (!res.ok()) {
+
+  Result lockResult = _trx->lockRecursive(cid, AccessMode::Type::READ);
+  
+  if (!lockResult.ok() && !lockResult.is(TRI_ERROR_LOCKED)) {
     LOG_TOPIC(ERR, arangodb::Logger::FIXME)
-        << "Logging Shard " << shard << " lead to exception '"
-        << res.errorNumber() << "' (" << res.errorMessage() << ") ";
+        << "Locking shard " << shard << " lead to exception '"
+        << lockResult.errorNumber() << "' (" << lockResult.errorMessage() << ") ";
     return false;
   }
+
   return true;
 }
 
diff --git a/arangod/GeneralServer/GeneralServerFeature.cpp b/arangod/GeneralServer/GeneralServerFeature.cpp
index 3d0340b3c6..9fc7f0d6c0 100644
--- a/arangod/GeneralServer/GeneralServerFeature.cpp
+++ b/arangod/GeneralServer/GeneralServerFeature.cpp
@@ -89,7 +89,6 @@
 #include "Ssl/SslServerFeature.h"
 #include "StorageEngine/EngineSelectorFeature.h"
 #include "StorageEngine/StorageEngine.h"
-#include "V8Server/V8DealerFeature.h"
 
 using namespace arangodb;
 using namespace arangodb::rest;
diff --git a/arangod/MMFiles/MMFilesCollectionKeys.cpp b/arangod/MMFiles/MMFilesCollectionKeys.cpp
index 825d95cb7e..45d53e35bb 100644
--- a/arangod/MMFiles/MMFilesCollectionKeys.cpp
+++ b/arangod/MMFiles/MMFilesCollectionKeys.cpp
@@ -43,9 +43,10 @@
 
 using namespace arangodb;
 
-MMFilesCollectionKeys::MMFilesCollectionKeys(TRI_vocbase_t* vocbase, std::string const& name,
+MMFilesCollectionKeys::MMFilesCollectionKeys(TRI_vocbase_t* vocbase, std::unique_ptr<CollectionGuard> guard,
                                              TRI_voc_tick_t blockerId, double ttl)
-    : CollectionKeys(vocbase, name, ttl),
+    : CollectionKeys(vocbase, ttl),
+      _guard(std::move(guard)),
       _ditch(nullptr),
       _resolver(vocbase),
       _blockerId(blockerId) {
@@ -53,8 +54,6 @@ MMFilesCollectionKeys::MMFilesCollectionKeys(TRI_vocbase_t* vocbase, std::string
 
   // prevent the collection from being unloaded while the export is ongoing
   // this may throw
-  _guard.reset(new arangodb::CollectionGuard(vocbase, _name.c_str(), false));
-
   _collection = _guard->collection();
   TRI_ASSERT(_collection != nullptr);
 }
@@ -95,7 +94,10 @@ void MMFilesCollectionKeys::create(TRI_voc_tick_t maxTick) {
   // copy all document tokens into the result under the read-lock
   {
     auto ctx = transaction::StandaloneContext::Create(_collection->vocbase());
-    SingleCollectionTransaction trx(ctx, _name, AccessMode::Type::READ);
+    SingleCollectionTransaction trx(ctx, _collection->cid(), AccessMode::Type::READ);
+
+    // already locked by _guard
+    trx.addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
 
     Result res = trx.begin();
     if (!res.ok()) {
diff --git a/arangod/MMFiles/MMFilesCollectionKeys.h b/arangod/MMFiles/MMFilesCollectionKeys.h
index bb8321b38b..277e11ed25 100644
--- a/arangod/MMFiles/MMFilesCollectionKeys.h
+++ b/arangod/MMFiles/MMFilesCollectionKeys.h
@@ -46,7 +46,7 @@ class MMFilesCollectionKeys final : public CollectionKeys {
   MMFilesCollectionKeys(MMFilesCollectionKeys const&) = delete;
   MMFilesCollectionKeys& operator=(MMFilesCollectionKeys const&) = delete;
 
-  MMFilesCollectionKeys(TRI_vocbase_t*, std::string const& name, 
+  MMFilesCollectionKeys(TRI_vocbase_t*, std::unique_ptr<CollectionGuard> guard,
                         TRI_voc_tick_t blockerId, double ttl);
 
   ~MMFilesCollectionKeys();
diff --git a/arangod/MMFiles/MMFilesCompactorThread.cpp b/arangod/MMFiles/MMFilesCompactorThread.cpp
index fd8280596d..c5cb0616f0 100644
--- a/arangod/MMFiles/MMFilesCompactorThread.cpp
+++ b/arangod/MMFiles/MMFilesCompactorThread.cpp
@@ -440,6 +440,9 @@ void MMFilesCompactorThread::compactDatafiles(LogicalCollection* collection,
   trx.addHint(transaction::Hints::Hint::NO_ABORT_MARKER);
   trx.addHint(transaction::Hints::Hint::NO_COMPACTION_LOCK);
   trx.addHint(transaction::Hints::Hint::NO_THROTTLING);
+  // when we get into this function, the caller has already acquired the
+  // collection's status lock - so we better do not lock it again
+  trx.addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
 
   CompactionInitialContext initial = getCompactionContext(&trx, collection, toCompact);
 
@@ -999,6 +1002,9 @@ uint64_t MMFilesCompactorThread::getNumberOfDocuments(LogicalCollection* collect
   // if lock acquisition fails, we go on and report an (arbitrary) positive number
   trx.addHint(transaction::Hints::Hint::TRY_LOCK); 
   trx.addHint(transaction::Hints::Hint::NO_THROTTLING);
+  // when we get into this function, the caller has already acquired the
+  // collection's status lock - so we better do not lock it again
+  trx.addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
 
   Result res = trx.begin();
 
diff --git a/arangod/MMFiles/MMFilesLogfileManager.cpp b/arangod/MMFiles/MMFilesLogfileManager.cpp
index 29caa8b147..0af2371004 100644
--- a/arangod/MMFiles/MMFilesLogfileManager.cpp
+++ b/arangod/MMFiles/MMFilesLogfileManager.cpp
@@ -704,7 +704,7 @@ MMFilesWalSlotInfo MMFilesLogfileManager::allocate(uint32_t size) {
 
 // allocate space in a logfile for later writing
 MMFilesWalSlotInfo MMFilesLogfileManager::allocate(TRI_voc_tick_t databaseId,
-                                  TRI_voc_cid_t collectionId, uint32_t size) {
+                                                   TRI_voc_cid_t collectionId, uint32_t size) {
   TRI_ASSERT(size >= sizeof(MMFilesMarker));
 
   if (!_allowWrites) {
@@ -768,10 +768,10 @@ MMFilesWalSlotInfoCopy MMFilesLogfileManager::allocateAndWrite(MMFilesWalMarker
 // memcpy the data into the WAL region and return the filled slot
 // to the WAL logfile manager
 MMFilesWalSlotInfoCopy MMFilesLogfileManager::writeSlot(MMFilesWalSlotInfo& slotInfo,
-                                       MMFilesWalMarker const* marker,
-                                       bool wakeUpSynchronizer,
-                                       bool waitForSyncRequested,
-                                       bool waitUntilSyncDone) {
+                                                        MMFilesWalMarker const* marker,
+                                                        bool wakeUpSynchronizer,
+                                                        bool waitForSyncRequested,
+                                                        bool waitUntilSyncDone) {
   TRI_ASSERT(slotInfo.slot != nullptr);
   TRI_ASSERT(marker != nullptr);
 
@@ -2207,6 +2207,9 @@ int MMFilesLogfileManager::inspectLogfiles() {
   // update the tick with the max tick we found in the WAL
   TRI_UpdateTickServer(_recoverState->lastTick);
 
+  // return the lock here to ensure proper locking order
+  writeLocker.unlock();
+
   TRI_ASSERT(_slots != nullptr);
   // set the last ticks we found in existing logfile data
   _slots->setLastTick(_recoverState->lastTick);
diff --git a/arangod/MMFiles/MMFilesRestReplicationHandler.cpp b/arangod/MMFiles/MMFilesRestReplicationHandler.cpp
index 6b2d066b7b..233a53e530 100644
--- a/arangod/MMFiles/MMFilesRestReplicationHandler.cpp
+++ b/arangod/MMFiles/MMFilesRestReplicationHandler.cpp
@@ -652,9 +652,9 @@ void MMFilesRestReplicationHandler::handleCommandCreateKeys() {
     return;
   }
 
-  arangodb::CollectionGuard guard(_vocbase, c->cid(), false);
+  auto guard = std::make_unique<arangodb::CollectionGuard>(_vocbase, c->cid(), false);
 
-  arangodb::LogicalCollection* col = guard.collection();
+  arangodb::LogicalCollection* col = guard->collection();
   TRI_ASSERT(col != nullptr);
 
   // turn off the compaction for the collection
@@ -665,10 +665,10 @@ void MMFilesRestReplicationHandler::handleCommandCreateKeys() {
   if (res != TRI_ERROR_NO_ERROR) {
     THROW_ARANGO_EXCEPTION(res);
   }
-
+  
   // initialize a container with the keys
   auto keys =
-      std::make_unique<MMFilesCollectionKeys>(_vocbase, col->name(), id, 300.0);
+      std::make_unique<MMFilesCollectionKeys>(_vocbase, std::move(guard), id, 300.0);
 
   std::string const idString(std::to_string(keys->id()));
 
diff --git a/arangod/MMFiles/MMFilesTransactionCollection.cpp b/arangod/MMFiles/MMFilesTransactionCollection.cpp
index 28ff8da957..a3db945e44 100644
--- a/arangod/MMFiles/MMFilesTransactionCollection.cpp
+++ b/arangod/MMFiles/MMFilesTransactionCollection.cpp
@@ -46,13 +46,17 @@ MMFilesTransactionCollection::MMFilesTransactionCollection(TransactionState* trx
 MMFilesTransactionCollection::~MMFilesTransactionCollection() {}
 
 /// @brief request a main-level lock for a collection
-int MMFilesTransactionCollection::lock() {
-  return lock(_accessType, 0);
-}
+/// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+/// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+/// returns any other error code otherwise
+int MMFilesTransactionCollection::lockRecursive() { return lockRecursive(_accessType, 0); }
 
 /// @brief request a lock for a collection
-int MMFilesTransactionCollection::lock(AccessMode::Type accessType,
-                                       int nestingLevel) {
+/// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+/// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+/// returns any other error code otherwise
+int MMFilesTransactionCollection::lockRecursive(AccessMode::Type accessType,
+                                                int nestingLevel) {
   if (AccessMode::isWriteOrExclusive(accessType) && !AccessMode::isWriteOrExclusive(_accessType)) {
     // wrong lock type
     return TRI_ERROR_INTERNAL;
@@ -67,8 +71,8 @@ int MMFilesTransactionCollection::lock(AccessMode::Type accessType,
 }
 
 /// @brief request an unlock for a collection
-int MMFilesTransactionCollection::unlock(AccessMode::Type accessType,
-                                         int nestingLevel) {
+int MMFilesTransactionCollection::unlockRecursive(AccessMode::Type accessType,
+                                                  int nestingLevel) {
   if (AccessMode::isWriteOrExclusive(accessType) && !AccessMode::isWriteOrExclusive(_accessType)) {
     // wrong lock type: write-unlock requested but collection is read-only
     return TRI_ERROR_INTERNAL;
@@ -95,6 +99,13 @@ bool MMFilesTransactionCollection::isLocked(AccessMode::Type accessType, int nes
   
 /// @brief check whether a collection is locked at all
 bool MMFilesTransactionCollection::isLocked() const {
+  if (CollectionLockState::_noLockHeaders != nullptr) {
+    std::string collName(_collection->name());
+    auto it = CollectionLockState::_noLockHeaders->find(collName);
+    if (it != CollectionLockState::_noLockHeaders->end()) {
+      return true;
+    }
+  }
   return (_lockType != AccessMode::Type::NONE);
 }
   
@@ -151,8 +162,8 @@ bool MMFilesTransactionCollection::canAccess(AccessMode::Type accessType) const
   }
 
   // check if access type matches
-  if (AccessMode::AccessMode::isWriteOrExclusive(accessType) && 
-      !AccessMode::AccessMode::isWriteOrExclusive(_accessType)) {
+  if (AccessMode::isWriteOrExclusive(accessType) && 
+      !AccessMode::isWriteOrExclusive(_accessType)) {
     // type doesn't match. probably also a mistake by the caller
     return false;
   }
@@ -161,8 +172,8 @@ bool MMFilesTransactionCollection::canAccess(AccessMode::Type accessType) const
 }
 
 int MMFilesTransactionCollection::updateUsage(AccessMode::Type accessType, int nestingLevel) {
-  if (AccessMode::AccessMode::isWriteOrExclusive(accessType) && 
-      !AccessMode::AccessMode::isWriteOrExclusive(_accessType)) {
+  if (AccessMode::isWriteOrExclusive(accessType) && 
+      !AccessMode::isWriteOrExclusive(_accessType)) {
     if (nestingLevel > 0) {
       // trying to write access a collection that is only marked with
       // read-access
@@ -226,7 +237,7 @@ int MMFilesTransactionCollection::use(int nestingLevel) {
   TRI_ASSERT(physical != nullptr);
 
   if (nestingLevel == 0 &&
-      AccessMode::AccessMode::isWriteOrExclusive(_accessType)) {
+      AccessMode::isWriteOrExclusive(_accessType)) {
     // read-lock the compaction lock
     if (!_transaction->hasHint(transaction::Hints::Hint::NO_COMPACTION_LOCK)) {
       if (!_compactionLocked) {
@@ -239,19 +250,23 @@ int MMFilesTransactionCollection::use(int nestingLevel) {
   bool shouldLock = _transaction->hasHint(transaction::Hints::Hint::LOCK_ENTIRELY);
 
   if (!shouldLock) {
-    shouldLock = (AccessMode::AccessMode::isWriteOrExclusive(_accessType) && !_transaction->hasHint(transaction::Hints::Hint::SINGLE_OPERATION));
+    shouldLock = (!AccessMode::isNone(_accessType) && !_transaction->hasHint(transaction::Hints::Hint::SINGLE_OPERATION));
   }
 
   if (shouldLock && !isLocked()) {
     // r/w lock the collection
     int res = doLock(_accessType, nestingLevel);
 
-    if (res != TRI_ERROR_NO_ERROR) {
+    if (res == TRI_ERROR_LOCKED) {
+      // TRI_ERROR_LOCKED is not an error, but it indicates that the lock operation has actually acquired the lock
+      // (and that the lock has not been held before)
+      res = TRI_ERROR_NO_ERROR;
+    } else if (res != TRI_ERROR_NO_ERROR) {
       return res;
     }
   }
   
-  if (AccessMode::AccessMode::isWriteOrExclusive(_accessType) && _originalRevision == 0) {
+  if (AccessMode::isWriteOrExclusive(_accessType) && _originalRevision == 0) {
     // store original revision at transaction start
     _originalRevision = physical->revision();
   }
@@ -269,7 +284,7 @@ void MMFilesTransactionCollection::unuse(int nestingLevel) {
   // the top level transaction releases all collections
   if (nestingLevel == 0 && _collection != nullptr) {
     if (!_transaction->hasHint(transaction::Hints::Hint::NO_COMPACTION_LOCK)) {
-      if (AccessMode::AccessMode::isWriteOrExclusive(_accessType) && _compactionLocked) {
+      if (AccessMode::isWriteOrExclusive(_accessType) && _compactionLocked) {
         auto physical = static_cast<MMFilesCollection*>(_collection->getPhysical());
         TRI_ASSERT(physical != nullptr);
         // read-unlock the compaction lock
@@ -294,6 +309,9 @@ void MMFilesTransactionCollection::release() {
 }
 
 /// @brief lock a collection
+/// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+/// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+/// returns any other error code otherwise
 int MMFilesTransactionCollection::doLock(AccessMode::Type type, int nestingLevel) {
   if (_transaction->hasHint(transaction::Hints::Hint::LOCK_NEVER)) {
     // never lock
@@ -339,7 +357,11 @@ int MMFilesTransactionCollection::doLock(AccessMode::Type type, int nestingLevel
 
   if (res == TRI_ERROR_NO_ERROR) {
     _lockType = type;
-  } else if (res == TRI_ERROR_LOCK_TIMEOUT && timeout >= 0.1) {
+    // not an error, but we use TRI_ERROR_LOCKED to indicate that we actually acquired the lock ourselves
+    return TRI_ERROR_LOCKED;
+  }
+
+  if (res == TRI_ERROR_LOCK_TIMEOUT && timeout >= 0.1) {
     LOG_TOPIC(WARN, Logger::QUERIES) << "timed out after " << timeout << " s waiting for " << AccessMode::typeString(type) << "-lock on collection '" << _collection->name() << "'";
   } else if (res == TRI_ERROR_DEADLOCK) {
     LOG_TOPIC(WARN, Logger::QUERIES) << "deadlock detected while trying to acquire " << AccessMode::typeString(type) << "-lock on collection '" << _collection->name() << "'";
diff --git a/arangod/MMFiles/MMFilesTransactionCollection.h b/arangod/MMFiles/MMFilesTransactionCollection.h
index 2283e1848e..35f35804b8 100644
--- a/arangod/MMFiles/MMFilesTransactionCollection.h
+++ b/arangod/MMFiles/MMFilesTransactionCollection.h
@@ -45,13 +45,19 @@ class MMFilesTransactionCollection final : public TransactionCollection {
   ~MMFilesTransactionCollection();
   
   /// @brief request a main-level lock for a collection
-  int lock() override;
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
+  int lockRecursive() override;
  
   /// @brief request a lock for a collection
-  int lock(AccessMode::Type, int nestingLevel) override;
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
+  int lockRecursive(AccessMode::Type, int nestingLevel) override;
 
   /// @brief request an unlock for a collection
-  int unlock(AccessMode::Type, int nestingLevel) override;
+  int unlockRecursive(AccessMode::Type, int nestingLevel) override;
 
   /// @brief check whether a collection is locked in a specific mode in a transaction
   bool isLocked(AccessMode::Type, int nestingLevel) const override;
@@ -74,6 +80,9 @@ class MMFilesTransactionCollection final : public TransactionCollection {
   
  private:
   /// @brief request a lock for a collection
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
   int doLock(AccessMode::Type, int nestingLevel);
 
   /// @brief request an unlock for a collection
diff --git a/arangod/Replication/Syncer.cpp b/arangod/Replication/Syncer.cpp
index 7a40f7fe64..6c64902dcf 100644
--- a/arangod/Replication/Syncer.cpp
+++ b/arangod/Replication/Syncer.cpp
@@ -526,6 +526,10 @@ Result Syncer::createCollection(TRI_vocbase_t* vocbase,
       }
       SingleCollectionTransaction trx(transaction::StandaloneContext::Create(vocbase),
                                       guard.collection()->cid(), AccessMode::Type::WRITE);
+    
+      // already locked by guard above
+      trx.addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
+
       Result res = trx.begin();
       if (!res.ok()) {
         return res;
@@ -626,6 +630,9 @@ Result Syncer::createIndex(VPackSlice const& slice) {
     SingleCollectionTransaction trx(transaction::StandaloneContext::Create(vocbase),
                                     guard.collection()->cid(), AccessMode::Type::WRITE);
 
+    // already locked by guard above
+    trx.addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
+
     Result res = trx.begin();
 
     if (!res.ok()) {
diff --git a/arangod/RestHandler/RestCollectionHandler.cpp b/arangod/RestHandler/RestCollectionHandler.cpp
index 32fc7e0e5e..942a310480 100644
--- a/arangod/RestHandler/RestCollectionHandler.cpp
+++ b/arangod/RestHandler/RestCollectionHandler.cpp
@@ -133,7 +133,7 @@ void RestCollectionHandler::handleCommandGet() {
             VPackObjectBuilder obj(&builder, true);
             obj->add("checksum", result.slice().get("checksum"));
             obj->add("revision", result.slice().get("revision"));
-            collectionRepresentation(builder, name, /*showProperties*/ false,
+            collectionRepresentation(builder, coll, /*showProperties*/ false,
                                      /*showFigures*/ false, /*showCount*/ false,
                                      /*aggregateCount*/ false);
           } else {
@@ -142,18 +142,18 @@ void RestCollectionHandler::handleCommandGet() {
           }
         } else if (sub == "figures") {
           // /_api/collection/<identifier>/figures
-          collectionRepresentation(builder, name, /*showProperties*/ true,
+          collectionRepresentation(builder, coll, /*showProperties*/ true,
                                    /*showFigures*/ true, /*showCount*/ true,
                                    /*aggregateCount*/ true);
         } else if (sub == "count") {
           // /_api/collection/<identifier>/count
           bool details = _request->parsedValue("details", false);
-          collectionRepresentation(builder, name, /*showProperties*/ true,
+          collectionRepresentation(builder, coll, /*showProperties*/ true,
                                    /*showFigures*/ false, /*showCount*/ true,
                                    /*aggregateCount*/ !details);
         } else if (sub == "properties") {
           // /_api/collection/<identifier>/properties
-          collectionRepresentation(builder, name, /*showProperties*/ true,
+          collectionRepresentation(builder, coll, /*showProperties*/ true,
                                    /*showFigures*/ false, /*showCount*/ false,
                                    /*aggregateCount*/ false);
         } else if (sub == "revision") {
@@ -166,7 +166,7 @@ void RestCollectionHandler::handleCommandGet() {
           }
           VPackObjectBuilder obj(&builder, true);
           obj->add("revision", VPackValue(StringUtils::itoa(revisionId)));
-          collectionRepresentation(builder, name, /*showProperties*/ true,
+          collectionRepresentation(builder, coll, /*showProperties*/ true,
                                    /*showFigures*/ false, /*showCount*/ false,
                                    /*aggregateCount*/ false);
 
@@ -179,7 +179,7 @@ void RestCollectionHandler::handleCommandGet() {
           }
 
           VPackObjectBuilder obj(&builder, true);  // need to open object
-          collectionRepresentation(builder, name, /*showProperties*/ true,
+          collectionRepresentation(builder, coll, /*showProperties*/ true,
                                    /*showFigures*/ false, /*showCount*/ false,
                                    /*aggregateCount*/ false);
           auto shards =
diff --git a/arangod/RestHandler/RestReplicationHandler.cpp b/arangod/RestHandler/RestReplicationHandler.cpp
index 4cd3302133..a1ccbf8bb1 100644
--- a/arangod/RestHandler/RestReplicationHandler.cpp
+++ b/arangod/RestHandler/RestReplicationHandler.cpp
@@ -1537,6 +1537,11 @@ int RestReplicationHandler::processRestoreIndexes(VPackSlice const& collection,
     auto ctx = transaction::StandaloneContext::Create(_vocbase);
     SingleCollectionTransaction trx(ctx, collection->cid(),
                                     AccessMode::Type::EXCLUSIVE);
+
+    // collection status lock was already acquired by collection guard 
+    // above
+    trx.addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
+    
     Result res = trx.begin();
 
     if (!res.ok()) {
diff --git a/arangod/RestHandler/RestTransactionHandler.cpp b/arangod/RestHandler/RestTransactionHandler.cpp
index 7e226b7b87..037746bfc5 100644
--- a/arangod/RestHandler/RestTransactionHandler.cpp
+++ b/arangod/RestHandler/RestTransactionHandler.cpp
@@ -21,16 +21,14 @@
 /// @author Jan Christoph Uhde
 ////////////////////////////////////////////////////////////////////////////////
 
-#include "Basics/WriteLocker.h"
-#include "Basics/ReadLocker.h"
 #include "RestTransactionHandler.h"
-
 #include "ApplicationFeatures/ApplicationServer.h"
-#include "VocBase/Methods/Transactions.h"
+#include "Basics/ReadLocker.h"
+#include "Basics/WriteLocker.h"
 #include "Rest/HttpRequest.h"
-#include "Basics/voc-errors.h"
 #include "V8Server/V8Context.h"
 #include "V8Server/V8DealerFeature.h"
+#include "VocBase/Methods/Transactions.h"
 
 #include <velocypack/Builder.h>
 #include <velocypack/velocypack-aliases.h>
@@ -42,13 +40,12 @@ using namespace arangodb::rest;
 RestTransactionHandler::RestTransactionHandler(GeneralRequest* request, GeneralResponse* response)
   : RestVocbaseBaseHandler(request, response)
   , _v8Context(nullptr)
-  , _lock()
-{}
+  , _lock() {}
 
-void RestTransactionHandler::returnContext(){
-    WRITE_LOCKER(writeLock, _lock);
-    V8DealerFeature::DEALER->exitContext(_v8Context);
-    _v8Context = nullptr;
+void RestTransactionHandler::returnContext() {
+  WRITE_LOCKER(writeLock, _lock);
+  V8DealerFeature::DEALER->exitContext(_v8Context);
+  _v8Context = nullptr;
 }
 
 RestStatus RestTransactionHandler::execute() {
@@ -58,7 +55,7 @@ RestStatus RestTransactionHandler::execute() {
   }
 
   auto slice = _request->payload();
-  if(!slice.isObject()){
+  if (!slice.isObject()) {
     generateError(Result(TRI_ERROR_BAD_PARAMETER, "could not acquire v8 context"));
     return RestStatus::DONE;
   }
@@ -77,14 +74,14 @@ RestStatus RestTransactionHandler::execute() {
   try {
     {
       WRITE_LOCKER(lock, _lock);
-      if(_canceled){
+      if (_canceled) {
         generateCanceled();
         return RestStatus::DONE;
       }
     }
 
     Result res = executeTransaction(_v8Context->_isolate, _lock, _canceled, slice , portType, result);
-    if (res.ok()){
+    if (res.ok()) {
       VPackSlice slice = result.slice();
       if (slice.isNone()) {
         generateOk(rest::ResponseCode::OK, VPackSlice::nullSlice());
@@ -111,7 +108,7 @@ bool RestTransactionHandler::cancel() {
   _canceled.store(true);
   auto isolate = _v8Context->_isolate;
   if (!v8::V8::IsExecutionTerminating(isolate)) {
-      v8::V8::TerminateExecution(isolate);
+    v8::V8::TerminateExecution(isolate);
   }
   return true;
 }
diff --git a/arangod/RestHandler/RestTransactionHandler.h b/arangod/RestHandler/RestTransactionHandler.h
index b7709d8e19..4cf2af9b21 100644
--- a/arangod/RestHandler/RestTransactionHandler.h
+++ b/arangod/RestHandler/RestTransactionHandler.h
@@ -24,6 +24,7 @@
 #ifndef ARANGOD_REST_HANDLER_REST_TRANSACTION_HANDLER_H
 #define ARANGOD_REST_HANDLER_REST_TRANSACTION_HANDLER_H 1
 
+#include "Basics/Common.h"
 #include "Basics/ReadWriteLock.h"
 #include "RestHandler/RestVocbaseBaseHandler.h"
 
diff --git a/arangod/RocksDBEngine/RocksDBReplicationContext.cpp b/arangod/RocksDBEngine/RocksDBReplicationContext.cpp
index 649fc64971..c69a1c4fcd 100644
--- a/arangod/RocksDBEngine/RocksDBReplicationContext.cpp
+++ b/arangod/RocksDBEngine/RocksDBReplicationContext.cpp
@@ -85,9 +85,9 @@ void RocksDBReplicationContext::bind(TRI_vocbase_t* vocbase) {
   if (!_trx || !_guard || (_guard->database() != vocbase)) {
     rocksdb::Snapshot const* snap = nullptr;
     if (_trx) {
-      _trx->abort();
       auto state = RocksDBTransactionState::toState(_trx.get());
       snap = state->stealSnapshot();
+      _trx->abort();
       _trx.reset();
     }
     
diff --git a/arangod/RocksDBEngine/RocksDBTransactionCollection.cpp b/arangod/RocksDBEngine/RocksDBTransactionCollection.cpp
index 60dc579f29..9696656410 100644
--- a/arangod/RocksDBEngine/RocksDBTransactionCollection.cpp
+++ b/arangod/RocksDBEngine/RocksDBTransactionCollection.cpp
@@ -50,11 +50,17 @@ RocksDBTransactionCollection::RocksDBTransactionCollection(
 RocksDBTransactionCollection::~RocksDBTransactionCollection() {}
 
 /// @brief request a main-level lock for a collection
-int RocksDBTransactionCollection::lock() { return lock(_accessType, 0); }
+/// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+/// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+/// returns any other error code otherwise
+int RocksDBTransactionCollection::lockRecursive() { return lockRecursive(_accessType, 0); }
 
 /// @brief request a lock for a collection
-int RocksDBTransactionCollection::lock(AccessMode::Type accessType,
-                                       int nestingLevel) {
+/// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+/// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+/// returns any other error code otherwise
+int RocksDBTransactionCollection::lockRecursive(AccessMode::Type accessType,
+                                                int nestingLevel) {
   if (AccessMode::isWriteOrExclusive(accessType) &&
       !AccessMode::isWriteOrExclusive(_accessType)) {
     // wrong lock type
@@ -70,8 +76,8 @@ int RocksDBTransactionCollection::lock(AccessMode::Type accessType,
 }
 
 /// @brief request an unlock for a collection
-int RocksDBTransactionCollection::unlock(AccessMode::Type accessType,
-                                         int nestingLevel) {
+int RocksDBTransactionCollection::unlockRecursive(AccessMode::Type accessType,
+                                                  int nestingLevel) {
   if (AccessMode::isWriteOrExclusive(accessType) &&
       !AccessMode::isWriteOrExclusive(_accessType)) {
     // wrong lock type: write-unlock requested but collection is read-only
@@ -102,6 +108,13 @@ bool RocksDBTransactionCollection::isLocked(AccessMode::Type accessType,
 
 /// @brief check whether a collection is locked at all
 bool RocksDBTransactionCollection::isLocked() const {
+  if (CollectionLockState::_noLockHeaders != nullptr) {
+    std::string collName(_collection->name());
+    auto it = CollectionLockState::_noLockHeaders->find(collName);
+    if (it != CollectionLockState::_noLockHeaders->end()) {
+      return true;
+    }
+  }
   return (_lockType != AccessMode::Type::NONE);
 }
 
@@ -200,8 +213,12 @@ int RocksDBTransactionCollection::use(int nestingLevel) {
   if (AccessMode::isWriteOrExclusive(_accessType) && !isLocked()) {
     // r/w lock the collection
     int res = doLock(_accessType, nestingLevel);
-
-    if (res != TRI_ERROR_NO_ERROR) {
+  
+    if (res == TRI_ERROR_LOCKED) {
+      // TRI_ERROR_LOCKED is not an error, but it indicates that the lock operation has actually acquired the lock
+      // (and that the lock has not been held before)
+      res = TRI_ERROR_NO_ERROR;
+    } else if (res != TRI_ERROR_NO_ERROR) {
       return res;
     }
   }
@@ -273,9 +290,13 @@ void RocksDBTransactionCollection::commitCounts() {
 }
 
 /// @brief lock a collection
+/// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+/// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+/// returns any other error code otherwise
 int RocksDBTransactionCollection::doLock(AccessMode::Type type,
                                          int nestingLevel) {
   if (!AccessMode::isWriteOrExclusive(type)) {
+    _lockType = type;
     return TRI_ERROR_NO_ERROR;
   }
 
@@ -323,7 +344,11 @@ int RocksDBTransactionCollection::doLock(AccessMode::Type type,
 
   if (res == TRI_ERROR_NO_ERROR) {
     _lockType = type;
-  } else if (res == TRI_ERROR_LOCK_TIMEOUT && timeout >= 0.1) {
+    // not an error, but we use TRI_ERROR_LOCKED to indicate that we actually acquired the lock ourselves
+    return TRI_ERROR_LOCKED;
+  } 
+  
+  if (res == TRI_ERROR_LOCK_TIMEOUT && timeout >= 0.1) {
     LOG_TOPIC(WARN, Logger::QUERIES)
         << "timed out after " << timeout << " s waiting for "
         << AccessMode::typeString(type) << "-lock on collection '"
@@ -338,6 +363,7 @@ int RocksDBTransactionCollection::doUnlock(AccessMode::Type type,
                                            int nestingLevel) {
   if (!AccessMode::isWriteOrExclusive(type) ||
       !AccessMode::isWriteOrExclusive(_lockType)) {
+    _lockType = AccessMode::Type::NONE;
     return TRI_ERROR_NO_ERROR;
   }
 
diff --git a/arangod/RocksDBEngine/RocksDBTransactionCollection.h b/arangod/RocksDBEngine/RocksDBTransactionCollection.h
index d10f11a790..e128add90c 100644
--- a/arangod/RocksDBEngine/RocksDBTransactionCollection.h
+++ b/arangod/RocksDBEngine/RocksDBTransactionCollection.h
@@ -44,13 +44,19 @@ class RocksDBTransactionCollection final : public TransactionCollection {
   ~RocksDBTransactionCollection();
 
   /// @brief request a main-level lock for a collection
-  int lock() override;
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
+  int lockRecursive() override;
 
   /// @brief request a lock for a collection
-  int lock(AccessMode::Type, int nestingLevel) override;
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
+  int lockRecursive(AccessMode::Type, int nestingLevel) override;
 
   /// @brief request an unlock for a collection
-  int unlock(AccessMode::Type, int nestingLevel) override;
+  int unlockRecursive(AccessMode::Type, int nestingLevel) override;
 
   /// @brief check whether a collection is locked in a specific mode in a
   /// transaction
@@ -86,6 +92,9 @@ class RocksDBTransactionCollection final : public TransactionCollection {
 
  private:
   /// @brief request a lock for a collection
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
   int doLock(AccessMode::Type, int nestingLevel);
 
   /// @brief request an unlock for a collection
diff --git a/arangod/RocksDBEngine/RocksDBTransactionState.cpp b/arangod/RocksDBEngine/RocksDBTransactionState.cpp
index 6f08075796..fcc70e3ac9 100644
--- a/arangod/RocksDBEngine/RocksDBTransactionState.cpp
+++ b/arangod/RocksDBEngine/RocksDBTransactionState.cpp
@@ -501,7 +501,7 @@ void RocksDBTransactionState::donateSnapshot(rocksdb::Snapshot const* snap) {
 rocksdb::Snapshot const* RocksDBTransactionState::stealSnapshot() {
   TRI_ASSERT(_snapshot != nullptr);
   TRI_ASSERT(isReadOnlyTransaction());
-  TRI_ASSERT(_status == transaction::Status::COMMITTED || _status == transaction::Status::ABORTED);
+  TRI_ASSERT(_status == transaction::Status::RUNNING);
   rocksdb::Snapshot const* snap = _snapshot;
   _snapshot = nullptr;
   return snap;
diff --git a/arangod/Scheduler/Scheduler.cpp b/arangod/Scheduler/Scheduler.cpp
index 38951750a0..d5203136ff 100644
--- a/arangod/Scheduler/Scheduler.cpp
+++ b/arangod/Scheduler/Scheduler.cpp
@@ -404,7 +404,7 @@ void Scheduler::rebalanceThreads() {
       uint64_t const nrWorking = numWorking(counters);
       uint64_t const nrBlocked = numBlocked(counters);
 
-      if (nrRunning >= std::max(_nrMinimum, nrWorking + nrQueued)) { 
+      if (nrRunning >= std::max(_nrMinimum, nrWorking + nrBlocked + nrQueued + 1)) { 
         // all threads are working, and none are blocked. so there is no
         // need to start a new thread now
         if (nrWorking == nrRunning) {
diff --git a/arangod/StorageEngine/TransactionCollection.h b/arangod/StorageEngine/TransactionCollection.h
index 142e3e7294..cf681ac3e6 100644
--- a/arangod/StorageEngine/TransactionCollection.h
+++ b/arangod/StorageEngine/TransactionCollection.h
@@ -58,13 +58,19 @@ class TransactionCollection {
   AccessMode::Type accessType() const { return _accessType; }
 
   /// @brief request a main-level lock for a collection
-  virtual int lock() = 0;
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
+  virtual int lockRecursive() = 0;
  
   /// @brief request a lock for a collection
-  virtual int lock(AccessMode::Type, int nestingLevel) = 0;
+  /// returns TRI_ERROR_LOCKED in case the lock was successfully acquired
+  /// returns TRI_ERROR_NO_ERROR in case the lock does not need to be acquired and no other error occurred
+  /// returns any other error code otherwise
+  virtual int lockRecursive(AccessMode::Type, int nestingLevel) = 0;
 
   /// @brief request an unlock for a collection
-  virtual int unlock(AccessMode::Type, int nestingLevel) = 0;
+  virtual int unlockRecursive(AccessMode::Type, int nestingLevel) = 0;
 
   /// @brief check whether a collection is locked in a specific mode in a transaction
   virtual bool isLocked(AccessMode::Type, int nestingLevel) const = 0;
diff --git a/arangod/StorageEngine/TransactionState.cpp b/arangod/StorageEngine/TransactionState.cpp
index 5983e32e83..3b9ae7680f 100644
--- a/arangod/StorageEngine/TransactionState.cpp
+++ b/arangod/StorageEngine/TransactionState.cpp
@@ -214,9 +214,9 @@ int TransactionState::unuseCollections(int nestingLevel) {
 
 int TransactionState::lockCollections() {
   for (auto& trxCollection : _collections) {
-    int res = trxCollection->lock();
+    int res = trxCollection->lockRecursive();
 
-    if (res != TRI_ERROR_NO_ERROR) {
+    if (res != TRI_ERROR_NO_ERROR && res != TRI_ERROR_LOCKED) {
       return res;
     }
   }
diff --git a/arangod/Transaction/Methods.cpp b/arangod/Transaction/Methods.cpp
index f6a7734a57..0b2463110e 100644
--- a/arangod/Transaction/Methods.cpp
+++ b/arangod/Transaction/Methods.cpp
@@ -819,18 +819,18 @@ OperationResult transaction::Methods::anyLocal(
   if (cid == 0) {
     throwCollectionNotFound(collectionName.c_str());
   }
-
+  
   pinData(cid);  // will throw when it fails
-
-  Result res = lock(cid, AccessMode::Type::READ);
-
-  if (!res.ok()) {
-    return OperationResult(res);
-  }
-
+  
   VPackBuilder resultBuilder;
   resultBuilder.openArray();
 
+  Result lockResult = lockRecursive(cid, AccessMode::Type::READ);
+
+  if (!lockResult.ok() && !lockResult.is(TRI_ERROR_LOCKED)) {
+    return OperationResult(lockResult);
+  }
+  
   ManagedDocumentResult mmdr;
   std::unique_ptr<OperationCursor> cursor =
       indexScan(collectionName, transaction::Methods::CursorType::ANY, &mmdr, false);
@@ -839,13 +839,15 @@ OperationResult transaction::Methods::anyLocal(
     resultBuilder.add(slice);
   });
 
-  resultBuilder.close();
+  if (lockResult.is(TRI_ERROR_LOCKED)) {
+    Result res = unlockRecursive(cid, AccessMode::Type::READ);
 
-  res = unlock(cid, AccessMode::Type::READ);
-
-  if (!res.ok()) {
-    return OperationResult(res);
+    if (!res.ok()) {
+      return OperationResult(res);
+    }
   }
+  
+  resultBuilder.close();
 
   return OperationResult(Result(), resultBuilder.steal(), _transactionContextPtr->orderCustomTypeHandler(), false);
 }
@@ -934,21 +936,25 @@ void transaction::Methods::invokeOnAllElements(
 
   TRI_voc_cid_t cid = addCollectionAtRuntime(collectionName);
   TransactionCollection* trxCol = trxCollection(cid, AccessMode::Type::READ);
-  LogicalCollection* logical = documentCollection(trxCol);
-  TRI_ASSERT(logical != nullptr);
-  _transactionContextPtr->pinData(logical);
+  LogicalCollection* collection = documentCollection(trxCol);
+  TRI_ASSERT(collection != nullptr);
+  _transactionContextPtr->pinData(collection);
 
-  Result res = trxCol->lock(AccessMode::Type::READ, _state->nestingLevel());
-  if (!res.ok()) {
-    THROW_ARANGO_EXCEPTION(res);
+  Result lockResult = trxCol->lockRecursive(AccessMode::Type::READ, _state->nestingLevel());
+  if (!lockResult.ok() && !lockResult.is(TRI_ERROR_LOCKED)) {
+    THROW_ARANGO_EXCEPTION(lockResult);
   }
+  
+  TRI_ASSERT(isLocked(collection, AccessMode::Type::READ));
 
-  logical->invokeOnAllElements(this, callback);
+  collection->invokeOnAllElements(this, callback);
 
-  res = trxCol->unlock(AccessMode::Type::READ, _state->nestingLevel());
+  if (lockResult.is(TRI_ERROR_LOCKED)) {
+    Result res = trxCol->unlockRecursive(AccessMode::Type::READ, _state->nestingLevel());
 
-  if (!res.ok()) {
-    THROW_ARANGO_EXCEPTION(res);
+    if (!res.ok()) {
+      THROW_ARANGO_EXCEPTION(res);
+    }
   }
 }
 
@@ -1708,12 +1714,12 @@ OperationResult transaction::Methods::modifyLocal(
 
   // Update/replace are a read and a write, let's get the write lock already
   // for the read operation:
-  Result res = lock(cid, AccessMode::Type::WRITE);
+  Result lockResult = lockRecursive(cid, AccessMode::Type::WRITE);
 
-  if (!res.ok()) {
-    return OperationResult(res);
+  if (!lockResult.ok() && !lockResult.is(TRI_ERROR_LOCKED)) {
+    return OperationResult(lockResult);
   }
-
+  
   VPackBuilder resultBuilder;  // building the complete result
   TRI_voc_tick_t maxTick = 0;
 
@@ -1775,6 +1781,7 @@ OperationResult transaction::Methods::modifyLocal(
 
   bool multiCase = newValue.isArray();
   std::unordered_map<int, size_t> errorCounter;
+  Result res;
   if (multiCase) {
     {
       VPackArrayBuilder guard(&resultBuilder);
@@ -2228,16 +2235,16 @@ OperationResult transaction::Methods::allLocal(
   TRI_voc_cid_t cid = addCollectionAtRuntime(collectionName);
 
   pinData(cid);  // will throw when it fails
-
-  Result res = lock(cid, AccessMode::Type::READ);
-
-  if (!res.ok()) {
-    return OperationResult(res);
-  }
-
+  
   VPackBuilder resultBuilder;
   resultBuilder.openArray();
 
+  Result lockResult = lockRecursive(cid, AccessMode::Type::READ);
+
+  if (!lockResult.ok() && !lockResult.is(TRI_ERROR_LOCKED)) {
+    return OperationResult(lockResult);
+  }
+
   ManagedDocumentResult mmdr;
   std::unique_ptr<OperationCursor> cursor =
       indexScan(collectionName, transaction::Methods::CursorType::ALL, &mmdr, false);
@@ -2251,13 +2258,15 @@ OperationResult transaction::Methods::allLocal(
   };
   cursor->allDocuments(cb);
 
-  resultBuilder.close();
+  if (lockResult.is(TRI_ERROR_LOCKED)) {
+    Result res = unlockRecursive(cid, AccessMode::Type::READ);
 
-  res = unlock(cid, AccessMode::Type::READ);
-
-  if (res.ok()) {
-    return OperationResult(res);
+    if (res.ok()) {
+      return OperationResult(res);
+    }
   }
+  
+  resultBuilder.close();
 
   return OperationResult(Result(), resultBuilder.steal(), _transactionContextPtr->orderCustomTypeHandler(), false);
 }
@@ -2317,19 +2326,25 @@ OperationResult transaction::Methods::truncateLocal(
 
   pinData(cid);  // will throw when it fails
 
-  Result res = lock(cid, AccessMode::Type::WRITE);
+  Result lockResult = lockRecursive(cid, AccessMode::Type::WRITE);
 
-  if (!res.ok()) {
-    return OperationResult(res);
+  if (!lockResult.ok() && !lockResult.is(TRI_ERROR_LOCKED)) {
+    return OperationResult(lockResult);
   }
+  
+  TRI_ASSERT(isLocked(collection, AccessMode::Type::WRITE));
 
   try {
     collection->truncate(this, options);
   } catch (basics::Exception const& ex) {
-    unlock(cid, AccessMode::Type::WRITE);
+    if (lockResult.is(TRI_ERROR_LOCKED)) {
+      unlockRecursive(cid, AccessMode::Type::WRITE);
+    }
     return OperationResult(Result(ex.code(), ex.what()));
   } catch (std::exception const& ex) {
-    unlock(cid, AccessMode::Type::WRITE);
+    if (lockResult.is(TRI_ERROR_LOCKED)) {
+      unlockRecursive(cid, AccessMode::Type::WRITE);
+    }
     return OperationResult(Result(TRI_ERROR_INTERNAL, ex.what()));
   }
 
@@ -2400,7 +2415,10 @@ OperationResult transaction::Methods::truncateLocal(
     }
   }
 
-  res = unlock(cid, AccessMode::Type::WRITE);
+  Result res;
+  if (lockResult.is(TRI_ERROR_LOCKED)) {
+    res = unlockRecursive(cid, AccessMode::Type::WRITE);
+  }
 
   return OperationResult(res);
 }
@@ -2480,18 +2498,22 @@ OperationResult transaction::Methods::countLocal(
   TRI_voc_cid_t cid = addCollectionAtRuntime(collectionName);
   LogicalCollection* collection = documentCollection(trxCollection(cid));
 
-  Result res = lock(cid, AccessMode::Type::READ);
+  Result lockResult = lockRecursive(cid, AccessMode::Type::READ);
 
-  if (!res.ok()) {
-    return OperationResult(res);
+  if (!lockResult.ok() && !lockResult.is(TRI_ERROR_LOCKED)) {
+    return OperationResult(lockResult);
   }
+  
+  TRI_ASSERT(isLocked(collection, AccessMode::Type::READ));
 
   uint64_t num = collection->numberDocuments(this);
 
-  res = unlock(cid, AccessMode::Type::READ);
+  if (lockResult.is(TRI_ERROR_LOCKED)) {
+    Result res = unlockRecursive(cid, AccessMode::Type::READ);
 
-  if (!res.ok()) {
-    return OperationResult(res);
+    if (!res.ok()) {
+      return OperationResult(res);
+    }
   }
 
   VPackBuilder resultBuilder;
@@ -2832,25 +2854,25 @@ bool transaction::Methods::isLocked(LogicalCollection* document,
 }
 
 /// @brief read- or write-lock a collection
-Result transaction::Methods::lock(TRI_voc_cid_t cid,
-                                  AccessMode::Type type) {
+Result transaction::Methods::lockRecursive(TRI_voc_cid_t cid,
+                                           AccessMode::Type type) {
   if (_state == nullptr || _state->status() != transaction::Status::RUNNING) {
     return Result(TRI_ERROR_TRANSACTION_INTERNAL, "transaction not running on lock");
   }
   TransactionCollection* trxColl = trxCollection(cid, type);
   TRI_ASSERT(trxColl != nullptr);
-  return trxColl->lock(type, _state->nestingLevel());
+  return Result(trxColl->lockRecursive(type, _state->nestingLevel()));
 }
 
 /// @brief read- or write-unlock a collection
-Result transaction::Methods::unlock(TRI_voc_cid_t cid,
-                                    AccessMode::Type type) {
+Result transaction::Methods::unlockRecursive(TRI_voc_cid_t cid,
+                                             AccessMode::Type type) {
   if (_state == nullptr || _state->status() != transaction::Status::RUNNING) {
     return Result(TRI_ERROR_TRANSACTION_INTERNAL, "transaction not running on unlock");
   }
   TransactionCollection* trxColl = trxCollection(cid, type);
   TRI_ASSERT(trxColl != nullptr);
-  return trxColl->unlock(type, _state->nestingLevel());
+  return Result(trxColl->unlockRecursive(type, _state->nestingLevel()));
 }
 
 /// @brief get list of indexes for a collection
diff --git a/arangod/Transaction/Methods.h b/arangod/Transaction/Methods.h
index be7c1f58f3..e04189ddc0 100644
--- a/arangod/Transaction/Methods.h
+++ b/arangod/Transaction/Methods.h
@@ -67,6 +67,7 @@ class BaseEngine;
 namespace transaction {
 class CallbackInvoker;
 class Context;
+struct Options;
 }
 
 /// @brief forward declarations
@@ -80,8 +81,6 @@ class TransactionState;
 class TransactionCollection;
 
 namespace transaction {
-struct Options;
-  
 #ifdef USE_ENTERPRISE
   #define ENTERPRISE_VIRT virtual
 #else
@@ -497,10 +496,10 @@ class Methods {
   Result addCollection(std::string const&, AccessMode::Type);
 
   /// @brief read- or write-lock a collection
-  ENTERPRISE_VIRT Result lock(TRI_voc_cid_t, AccessMode::Type);
+  ENTERPRISE_VIRT Result lockRecursive(TRI_voc_cid_t, AccessMode::Type);
 
   /// @brief read- or write-unlock a collection
-  ENTERPRISE_VIRT Result unlock(TRI_voc_cid_t, AccessMode::Type);
+  ENTERPRISE_VIRT Result unlockRecursive(TRI_voc_cid_t, AccessMode::Type);
 
  private:
 
diff --git a/arangod/Transaction/V8Context.cpp b/arangod/Transaction/V8Context.cpp
index f197bdde5d..bd0bdfd733 100644
--- a/arangod/Transaction/V8Context.cpp
+++ b/arangod/Transaction/V8Context.cpp
@@ -23,6 +23,7 @@
 
 #include "V8Context.h"
 #include "StorageEngine/TransactionState.h"
+#include "Transaction/StandaloneContext.h"
 #include "Utils/CollectionNameResolver.h"
 
 #include <v8.h>
@@ -137,3 +138,11 @@ std::shared_ptr<transaction::V8Context> transaction::V8Context::Create(
     TRI_vocbase_t* vocbase, bool embeddable) {
   return std::make_shared<transaction::V8Context>(vocbase, embeddable);
 }
+      
+std::shared_ptr<transaction::Context> transaction::V8Context::CreateWhenRequired(
+    TRI_vocbase_t* vocbase, bool embeddable) {
+  if (v8::Isolate::GetCurrent() != nullptr) {
+    return Create(vocbase, embeddable);
+  }
+  return transaction::StandaloneContext::Create(vocbase);
+}
diff --git a/arangod/Transaction/V8Context.h b/arangod/Transaction/V8Context.h
index 4024af298f..9ac2fa24ff 100644
--- a/arangod/Transaction/V8Context.h
+++ b/arangod/Transaction/V8Context.h
@@ -75,8 +75,12 @@ class V8Context final : public Context {
   /// @brief check whether the transaction is embedded
   static bool isEmbedded();
   
-  /// @brief create a context, returned in a shared ptr
-  static std::shared_ptr<transaction::V8Context> Create(TRI_vocbase_t*, bool);
+  /// @brief create a context
+  static std::shared_ptr<transaction::V8Context> Create(TRI_vocbase_t*, bool embeddable);
+
+  /// @brief create a V8 transaction context if we are in a V8 isolate, and a standlone
+  /// transaction context otherwise
+  static std::shared_ptr<transaction::Context> CreateWhenRequired(TRI_vocbase_t*, bool embeddable);
 
  private:
 
diff --git a/arangod/Utils/CollectionGuard.h b/arangod/Utils/CollectionGuard.h
index d086565ea2..6286307184 100644
--- a/arangod/Utils/CollectionGuard.h
+++ b/arangod/Utils/CollectionGuard.h
@@ -121,6 +121,14 @@ class CollectionGuard {
   }
 
  public:
+  /// @brief prematurely release the usage lock
+  void release() {
+    if (_collection != nullptr) {
+      _vocbase->releaseCollection(_collection);
+      _collection = nullptr;
+    }
+  }
+
   /// @brief return the collection pointer
   inline arangodb::LogicalCollection* collection() const { return _collection; }
 
diff --git a/arangod/Utils/CollectionKeys.cpp b/arangod/Utils/CollectionKeys.cpp
index a2a3e5acd4..201b20c6fb 100644
--- a/arangod/Utils/CollectionKeys.cpp
+++ b/arangod/Utils/CollectionKeys.cpp
@@ -26,10 +26,9 @@
 
 using namespace arangodb;
 
-CollectionKeys::CollectionKeys(TRI_vocbase_t* vocbase, std::string const& name, double ttl)
+CollectionKeys::CollectionKeys(TRI_vocbase_t* vocbase, double ttl)
     : _vocbase(vocbase),
       _collection(nullptr),
-      _name(name),
       _id(0),
       _ttl(ttl),
       _expires(0.0),
diff --git a/arangod/Utils/CollectionKeys.h b/arangod/Utils/CollectionKeys.h
index 24314a257c..b264a039c6 100644
--- a/arangod/Utils/CollectionKeys.h
+++ b/arangod/Utils/CollectionKeys.h
@@ -47,8 +47,7 @@ class CollectionKeys {
   CollectionKeys(CollectionKeys const&) = delete;
   CollectionKeys& operator=(CollectionKeys const&) = delete;
 
-  CollectionKeys(TRI_vocbase_t*, std::string const& name, 
-                 double ttl);
+  CollectionKeys(TRI_vocbase_t*, double ttl);
 
   virtual ~CollectionKeys() = default;
 
@@ -110,7 +109,6 @@ class CollectionKeys {
  protected:
   TRI_vocbase_t* _vocbase;
   arangodb::LogicalCollection* _collection;
-  std::string const _name;
   CollectionKeysId _id;
   double _ttl;
   double _expires;
diff --git a/arangod/V8Server/V8Context.cpp b/arangod/V8Server/V8Context.cpp
index af831128dc..6a39d8941e 100644
--- a/arangod/V8Server/V8Context.cpp
+++ b/arangod/V8Server/V8Context.cpp
@@ -57,17 +57,15 @@ void V8Context::lockAndEnter() {
   TRI_ASSERT(_locker == nullptr);
   _locker = new v8::Locker(_isolate);
   _isolate->Enter();
-      
-  TRI_ASSERT(_locker->IsLocked(_isolate));
-  TRI_ASSERT(v8::Locker::IsLocked(_isolate));
+     
+  assertLocked(); 
 
   ++_invocations;
   ++_invocationsSinceLastGc;
 }
 
 void V8Context::unlockAndExit() {
-  TRI_ASSERT(_locker != nullptr);
-  TRI_ASSERT(_isolate != nullptr);
+  assertLocked();
   
   _isolate->Exit();
   delete _locker;
@@ -75,6 +73,13 @@ void V8Context::unlockAndExit() {
   
   TRI_ASSERT(!v8::Locker::IsLocked(_isolate));
 }
+
+void V8Context::assertLocked() const {
+  TRI_ASSERT(_locker != nullptr);
+  TRI_ASSERT(_isolate != nullptr);
+  TRI_ASSERT(_locker->IsLocked(_isolate));
+  TRI_ASSERT(v8::Locker::IsLocked(_isolate));
+}
   
 bool V8Context::hasGlobalMethodsQueued() {
   MUTEX_LOCKER(mutexLocker, _globalMethodsLock);
diff --git a/arangod/V8Server/V8Context.h b/arangod/V8Server/V8Context.h
index 6e6e3e899b..a4f1185b04 100644
--- a/arangod/V8Server/V8Context.h
+++ b/arangod/V8Server/V8Context.h
@@ -114,7 +114,7 @@ class V8Context {
 
   size_t id() const { return _id; }
   bool isDefault() const { return _id == 0; }
-  bool isUsed() const { return _locker != nullptr; }
+  void assertLocked() const;
   double age() const;
   void lockAndEnter();
   void unlockAndExit();
@@ -125,7 +125,6 @@ class V8Context {
   void setCleaned(double stamp);
 
   size_t const _id;
-
   v8::Persistent<v8::Context> _context;
   v8::Isolate* _isolate;
   v8::Locker* _locker;
@@ -147,6 +146,8 @@ class V8Context {
 class V8ContextGuard {
  public:
   explicit V8ContextGuard(V8Context* context);
+  V8ContextGuard(V8ContextGuard const&) = delete;
+  V8ContextGuard& operator=(V8ContextGuard const&) = delete;
   ~V8ContextGuard();
 
  private:
diff --git a/arangod/V8Server/V8DealerFeature.cpp b/arangod/V8Server/V8DealerFeature.cpp
index 301efb58c4..5402d530f5 100644
--- a/arangod/V8Server/V8DealerFeature.cpp
+++ b/arangod/V8Server/V8DealerFeature.cpp
@@ -101,11 +101,10 @@ V8DealerFeature::V8DealerFeature(
       _nrInflightContexts(0),
       _maxContextInvocations(0),
       _allowAdminExecute(false),
-      _ok(false),
       _nextId(0),
       _stopping(false),
       _gcFinished(false),
-      _contextsModificationBlockers(0) {
+      _dynamicContextCreationBlockers(0) {
   setOptional(false);
   requiresElevatedPrivileges(false);
   startsAfter("Action");
@@ -258,7 +257,7 @@ void V8DealerFeature::start() {
     CONDITION_LOCKER(guard, _contextCondition);
     _contexts.reserve(static_cast<size_t>(_nrMaxContexts));
     _busyContexts.reserve(static_cast<size_t>(_nrMaxContexts));
-    _freeContexts.reserve(static_cast<size_t>(_nrMaxContexts));
+    _idleContexts.reserve(static_cast<size_t>(_nrMaxContexts));
     _dirtyContexts.reserve(static_cast<size_t>(_nrMaxContexts));
   
     for (size_t i = 0; i < _nrMinContexts; ++i) {
@@ -277,7 +276,7 @@ void V8DealerFeature::start() {
       // apply context update is only run on contexts that no other
       // threads can see (yet)
       applyContextUpdate(context);
-      _freeContexts.push_back(context);
+      _idleContexts.push_back(context);
     }
   }
 
@@ -367,7 +366,7 @@ void V8DealerFeature::collectGarbage() {
   uint64_t const reducedWaitTime =
       static_cast<uint64_t>(_gcFrequency * 1000.0 * 200.0);
 
-  while (_stopping == 0) {
+  while (!_stopping) {
     try {
       V8Context* context = nullptr;
       bool wasDirty = false;
@@ -384,7 +383,7 @@ void V8DealerFeature::collectGarbage() {
           gotSignal = guard.wait(waitTime);
         }
 
-        if (preferFree && !_freeContexts.empty()) {
+        if (preferFree && !_idleContexts.empty()) {
           context = pickFreeContextForGc();
         }
 
@@ -394,7 +393,7 @@ void V8DealerFeature::collectGarbage() {
           if (context->invocationsSinceLastGc() < 50 && !context->_hasActiveExternals) {
             // don't collect this one yet. it doesn't have externals, so there
             // is no urge for garbage collection
-            _freeContexts.emplace_back(context);
+            _idleContexts.emplace_back(context);
             context = nullptr;
           } else {
             wasDirty = true;
@@ -402,7 +401,7 @@ void V8DealerFeature::collectGarbage() {
         }
 
         if (context == nullptr && !preferFree && !gotSignal &&
-            !_freeContexts.empty()) {
+            !_idleContexts.empty()) {
           // we timed out waiting for a signal, so we have idle time that we can
           // spend on running the GC pro-actively
           // We'll pick one of the free contexts and clean it up
@@ -444,8 +443,7 @@ void V8DealerFeature::collectGarbage() {
           {
             v8::Context::Scope contextScope(localContext);
 
-            TRI_ASSERT(context->_locker->IsLocked(isolate));
-            TRI_ASSERT(v8::Locker::IsLocked(isolate));
+            context->assertLocked();
 
             TRI_GET_GLOBALS();
             TRI_RunGarbageCollectionV8(isolate, 1.0);
@@ -464,7 +462,7 @@ void V8DealerFeature::collectGarbage() {
           if (_contexts.size() > _nrMinContexts && 
               !context->isDefault() &&
               context->shouldBeRemoved(_maxContextAge, _maxContextInvocations) &&
-              _contextsModificationBlockers == 0) {
+              _dynamicContextCreationBlockers == 0) {
             // remove the extra context as it is not needed anymore
             _contexts.erase(std::remove_if(_contexts.begin(), _contexts.end(), [&context](V8Context* c) {
               return (c->id() == context->id());
@@ -477,9 +475,9 @@ void V8DealerFeature::collectGarbage() {
           } else {
             // put it back into the free list
             if (wasDirty) {
-              _freeContexts.emplace_back(context);
+              _idleContexts.emplace_back(context);
             } else {
-              _freeContexts.insert(_freeContexts.begin(), context);
+              _idleContexts.insert(_idleContexts.begin(), context);
             }
             guard.broadcast();
           }
@@ -496,11 +494,11 @@ void V8DealerFeature::collectGarbage() {
   _gcFinished = true;
 }
   
-void V8DealerFeature::unblockContextsModification() {
+void V8DealerFeature::unblockDynamicContextCreation() {
   CONDITION_LOCKER(guard, _contextCondition);
     
-  TRI_ASSERT(_contextsModificationBlockers > 0);
-  --_contextsModificationBlockers;
+  TRI_ASSERT(_dynamicContextCreationBlockers > 0);
+  --_dynamicContextCreationBlockers;
 }
 
 void V8DealerFeature::loadJavaScriptFileInAllContexts(TRI_vocbase_t* vocbase,
@@ -517,27 +515,69 @@ void V8DealerFeature::loadJavaScriptFileInAllContexts(TRI_vocbase_t* vocbase,
   {
     CONDITION_LOCKER(guard, _contextCondition);
     
-    // block the addition or removal of contexts
-    ++_contextsModificationBlockers;
-  
+    while (_nrInflightContexts > 0) {
+      // wait until all pending context creation requests have been satisified
+      guard.wait(10000);
+    }
+
     // copy the list of contexts into a local variable
     contexts = _contexts;
+    // block the addition or removal of contexts
+    ++_dynamicContextCreationBlockers;
   }
 
-  TRI_DEFER(unblockContextsModification());
+  TRI_DEFER(unblockDynamicContextCreation());
+
+  LOG_TOPIC(TRACE, Logger::V8) << "loading JavaScript file '" << file << "' in all (" << contexts.size() << ") V8 context";
 
   // now safely scan the local copy of the contexts  
   for (auto& context : contexts) {
     CONDITION_LOCKER(guard, _contextCondition);
 
-    while (context->isUsed()) {
+    while (_busyContexts.find(context) != _busyContexts.end()) {
       // we must not enter the context if another thread is also using it...
       guard.wait(10000);
     }
 
-    TRI_ASSERT(!context->isUsed());
-    loadJavaScriptFileInContext(vocbase, file, context, builder);
-    TRI_ASSERT(!context->isUsed());
+    auto it = std::find(_dirtyContexts.begin(), _dirtyContexts.end(), context);
+    if (it != _dirtyContexts.end()) {
+      // context is in _dirtyContexts
+      // remove it from there
+      _dirtyContexts.erase(it);
+      
+      guard.unlock();
+      try {
+        loadJavaScriptFileInContext(vocbase, file, context, builder);
+      } catch (...) {
+        guard.lock();
+        _dirtyContexts.push_back(context);
+        throw;
+      }
+      // and re-insert it after we are done
+      guard.lock();
+      _dirtyContexts.push_back(context);
+    } else {
+      // if the context is neither busy nor dirty, it must be idle
+      auto it = std::find(_idleContexts.begin(), _idleContexts.end(), context);
+      if (it != _idleContexts.end()) {
+        // remove it from there
+        _idleContexts.erase(it);
+
+        guard.unlock();
+        try {
+          loadJavaScriptFileInContext(vocbase, file, context, builder);
+        } catch (...) {
+          guard.lock();
+          _idleContexts.push_back(context);
+          throw;
+        }
+        // and re-insert it after we are done
+        guard.lock();
+        _idleContexts.push_back(context);
+      } else {
+        LOG_TOPIC(WARN, Logger::V8) << "v8 context #" << context->id() << " has disappeared";
+      }
+    }
   }
 
   if (builder != nullptr) {
@@ -577,22 +617,13 @@ void V8DealerFeature::startGarbageCollection() {
   _gcFinished = false;
 }
 
-void V8DealerFeature::enterContextInternal(TRI_vocbase_t* vocbase,
+void V8DealerFeature::prepareLockedContext(TRI_vocbase_t* vocbase,
                                            V8Context* context,
                                            bool allowUseDatabase) {
-  context->lockAndEnter();
-  enterLockedContext(vocbase, context, allowUseDatabase);
-}
-
-void V8DealerFeature::enterLockedContext(TRI_vocbase_t* vocbase,
-                                         V8Context* context,
-                                         bool allowUseDatabase) {
   TRI_ASSERT(vocbase != nullptr);
 
   // when we get here, we should have a context and an isolate
-  TRI_ASSERT(context != nullptr);
-  TRI_ASSERT(context->_isolate != nullptr);
-  TRI_ASSERT(context->isUsed());
+  context->assertLocked();
 
   auto isolate = context->_isolate;
 
@@ -604,8 +635,7 @@ void V8DealerFeature::enterLockedContext(TRI_vocbase_t* vocbase,
     {
       v8::Context::Scope contextScope(localContext);
 
-      TRI_ASSERT(context->_locker->IsLocked(isolate));
-      TRI_ASSERT(v8::Locker::IsLocked(isolate));
+      context->assertLocked();
       TRI_GET_GLOBALS();
 
       // initialize the context data
@@ -663,10 +693,10 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
           break;
         }
 
-        for (auto it = _freeContexts.begin(); it != _freeContexts.end(); ++it) {
+        for (auto it = _idleContexts.begin(); it != _idleContexts.end(); ++it) {
           if ((*it)->id() == id) {
             context = (*it);
-            _freeContexts.erase(it);
+            _idleContexts.erase(it);
             _busyContexts.emplace(context);
             break;
           }
@@ -687,7 +717,6 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
         if (context != nullptr) {
           // found the context
           TRI_ASSERT(guard.isLocked());
-          context->lockAndEnter();
           break;
         }
 
@@ -702,7 +731,7 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
 
         if (!found) {
           vocbase->release();
-          LOG_TOPIC(WARN, arangodb::Logger::V8) << "specified context #" << id << " not found";
+          LOG_TOPIC(WARN, arangodb::Logger::V8) << "specified V8 context #" << id << " not found";
           return nullptr;
         }
       }
@@ -721,15 +750,14 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
   else {
     CONDITION_LOCKER(guard, _contextCondition);
 
-    while (_freeContexts.empty() && !_stopping) {
+    while (_idleContexts.empty() && !_stopping) {
       TRI_ASSERT(guard.isLocked());
 
       LOG_TOPIC(TRACE, arangodb::Logger::V8) << "waiting for unused V8 context";
 
       if (!_dirtyContexts.empty()) {
         // we'll use a dirty context in this case
-        V8Context* context = _dirtyContexts.back();
-        _freeContexts.push_back(context);
+        _idleContexts.push_back(_dirtyContexts.back());
         _dirtyContexts.pop_back();
         break;
       }
@@ -739,7 +767,7 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
          (forceContext == ANY_CONTEXT_OR_PRIORITY && (_contexts.size() + _nrInflightContexts <= _nrMaxContexts)));
       
       if (contextLimitNotExceeded &&
-          _contextsModificationBlockers == 0 && 
+          _dynamicContextCreationBlockers == 0 && 
           !MaxMapCountFeature::isNearMaxMappings()) {
   
         ++_nrInflightContexts;
@@ -753,6 +781,7 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
         } catch (...) {
           guard.lock();
 
+          // clean up state
           --_nrInflightContexts;
           throw;
         }
@@ -771,8 +800,9 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
           continue;
         }
 
+        TRI_ASSERT(guard.isLocked());
         try {
-          _freeContexts.push_back(context);
+          _idleContexts.push_back(context);
           LOG_TOPIC(DEBUG, Logger::V8) << "created additional V8 context #" << context->id() << ", number of contexts is now " << _contexts.size();
         } catch (...) {
           TRI_ASSERT(!_contexts.empty());
@@ -808,49 +838,39 @@ V8Context* V8DealerFeature::enterContext(TRI_vocbase_t* vocbase,
       return nullptr;
     }
 
-    TRI_ASSERT(!_freeContexts.empty());
+    TRI_ASSERT(!_idleContexts.empty());
 
-    context = _freeContexts.back();
+    context = _idleContexts.back();
     LOG_TOPIC(TRACE, arangodb::Logger::V8) << "found unused V8 context #" << context->id();
     TRI_ASSERT(context != nullptr);
 
-    _freeContexts.pop_back();
+    _idleContexts.pop_back();
 
     // should not fail because we reserved enough space beforehand
     _busyContexts.emplace(context);
-
-    context->lockAndEnter();
   }
   
   TRI_ASSERT(context != nullptr);
-  TRI_ASSERT(context->isUsed());
+  context->lockAndEnter();
+  context->assertLocked();
 
-  enterLockedContext(vocbase, context, allowUseDatabase);
+  prepareLockedContext(vocbase, context, allowUseDatabase);
   return context;
 }
 
 void V8DealerFeature::exitContextInternal(V8Context* context) {
-  try {
-    exitLockedContext(context);
-    context->unlockAndExit();
-  } catch (...) {
-    // make sure the context will be exited
-    context->unlockAndExit();
-    throw;
-  }
+  TRI_DEFER(context->unlockAndExit());
+  cleanupLockedContext(context);
 }
 
-void V8DealerFeature::exitLockedContext(V8Context* context) {
+void V8DealerFeature::cleanupLockedContext(V8Context* context) {
   TRI_ASSERT(context != nullptr);
 
   LOG_TOPIC(TRACE, arangodb::Logger::V8) << "leaving V8 context #" << context->id();
 
   auto isolate = context->_isolate;
   TRI_ASSERT(isolate != nullptr);
-  TRI_ASSERT(context->_locker != nullptr);
-  TRI_ASSERT(context->_locker->IsLocked(isolate));
-  TRI_ASSERT(v8::Locker::IsLocked(isolate));
-  TRI_ASSERT(context->isUsed());
+  context->assertLocked();
 
   bool canceled = false;
 
@@ -903,8 +923,7 @@ void V8DealerFeature::exitLockedContext(V8Context* context) {
 
     // run global context methods
     if (runGlobal) {
-      TRI_ASSERT(context->_locker->IsLocked(isolate));
-      TRI_ASSERT(v8::Locker::IsLocked(isolate));
+      context->assertLocked();
 
       try {
         context->handleGlobalContextMethods();
@@ -927,7 +946,7 @@ void V8DealerFeature::exitLockedContext(V8Context* context) {
 }
 
 void V8DealerFeature::exitContext(V8Context* context) {
-  exitLockedContext(context);
+  cleanupLockedContext(context);
 
   V8GcThread* gc = static_cast<V8GcThread*>(_gcThread.get());
   
@@ -956,10 +975,10 @@ void V8DealerFeature::exitContext(V8Context* context) {
       performGarbageCollection = true;
     }
 
-    CONDITION_LOCKER(guard, _contextCondition);
     context->unlockAndExit();
+    CONDITION_LOCKER(guard, _contextCondition);
 
-    if (performGarbageCollection && (forceGarbageCollection || !_freeContexts.empty())) {
+    if (performGarbageCollection && (forceGarbageCollection || !_idleContexts.empty())) {
       // only add the context to the dirty list if there is at least one other
       // free context
 
@@ -969,7 +988,7 @@ void V8DealerFeature::exitContext(V8Context* context) {
     } else {
       // note that re-adding the context here should not fail as we reserved
       // enough room for all contexts during startup
-      _freeContexts.emplace_back(context);
+      _idleContexts.emplace_back(context);
     }
 
     _busyContexts.erase(context);
@@ -977,13 +996,13 @@ void V8DealerFeature::exitContext(V8Context* context) {
     LOG_TOPIC(TRACE, arangodb::Logger::V8) << "returned dirty V8 context #" << context->id();
     guard.broadcast();
   } else {
-    CONDITION_LOCKER(guard, _contextCondition);
     context->unlockAndExit();
+    CONDITION_LOCKER(guard, _contextCondition);
 
     _busyContexts.erase(context);
     // note that re-adding the context here should not fail as we reserved
     // enough room for all contexts during startup
-    _freeContexts.emplace_back(context);
+    _idleContexts.emplace_back(context);
 
     LOG_TOPIC(TRACE, arangodb::Logger::V8) << "returned dirty V8 context #" << context->id() << " back into free";
     guard.broadcast();
@@ -1011,7 +1030,8 @@ void V8DealerFeature::applyContextUpdate(V8Context* context) {
       continue;
     }
 
-    enterContextInternal(vocbase, context, true);
+    context->lockAndEnter();
+    prepareLockedContext(vocbase, context, true);
     TRI_DEFER(exitContextInternal(context));
 
     {
@@ -1091,7 +1111,7 @@ void V8DealerFeature::shutdownContexts() {
       std::this_thread::sleep_for(std::chrono::microseconds(10000));
     }
 
-    LOG_TOPIC(DEBUG, arangodb::Logger::V8) << "commanding GC Thread to terminate";
+    LOG_TOPIC(DEBUG, arangodb::Logger::V8) << "commanding V8 GC thread to terminate";
   }
 
   // shutdown all instances
@@ -1109,7 +1129,7 @@ void V8DealerFeature::shutdownContexts() {
 }
 
 V8Context* V8DealerFeature::pickFreeContextForGc() {
-  int const n = (int)_freeContexts.size();
+  int const n = static_cast<int>(_idleContexts.size());
 
   if (n == 0) {
     // this is easy...
@@ -1126,15 +1146,15 @@ V8Context* V8DealerFeature::pickFreeContextForGc() {
 
   for (int i = n - 1; i > 0; --i) {
     // check if there's actually anything to clean up in the context
-    if (_freeContexts[i]->invocationsSinceLastGc() < 50 &&
-        !_freeContexts[i]->_hasActiveExternals) {
+    if (_idleContexts[i]->invocationsSinceLastGc() < 50 &&
+        !_idleContexts[i]->_hasActiveExternals) {
       continue;
     }
 
     // compare last GC stamp
     if (pickedContextNr == -1 ||
-        _freeContexts[i]->_lastGcStamp <=
-            _freeContexts[pickedContextNr]->_lastGcStamp) {
+        _idleContexts[i]->_lastGcStamp <=
+            _idleContexts[pickedContextNr]->_lastGcStamp) {
       pickedContextNr = i;
     }
   }
@@ -1147,7 +1167,7 @@ V8Context* V8DealerFeature::pickFreeContextForGc() {
   }
 
   // this is the context to clean up
-  V8Context* context = _freeContexts[pickedContextNr];
+  V8Context* context = _idleContexts[pickedContextNr];
   TRI_ASSERT(context != nullptr);
 
   // now compare its last GC timestamp with the last global GC stamp
@@ -1161,10 +1181,10 @@ V8Context* V8DealerFeature::pickFreeContextForGc() {
   // around
   if (n > 1) {
     for (int i = pickedContextNr; i < n - 1; ++i) {
-      _freeContexts[i] = _freeContexts[i + 1];
+      _idleContexts[i] = _idleContexts[i + 1];
     }
   }
-  _freeContexts.pop_back();
+  _idleContexts.pop_back();
 
   return context;
 }
@@ -1187,7 +1207,7 @@ V8Context* V8DealerFeature::buildContext(size_t id) {
     // and automatically exit and unlock it when it runs out of scope
     V8ContextGuard contextGuard(context.get());
 
-    v8::HandleScope handleScope(isolate);
+    v8::HandleScope scope(isolate);
 
     v8::Handle<v8::ObjectTemplate> global = v8::ObjectTemplate::New(isolate);
 
@@ -1255,7 +1275,7 @@ V8Context* V8DealerFeature::buildContext(size_t id) {
                                            v8::ReadOnly);
         }
 
-        for (auto j : _definedStrings) {
+        for (auto const& j : _definedStrings) {
           localContext->Global()->ForceSet(TRI_V8_STD_STRING(isolate, j.first),
                                            TRI_V8_STD_STRING(isolate, j.second),
                                            v8::ReadOnly);
@@ -1291,7 +1311,7 @@ V8DealerFeature::stats V8DealerFeature::getCurrentContextNumbers() {
     _contexts.size(),
     _busyContexts.size(),
     _dirtyContexts.size(),
-    _freeContexts.size(),
+    _idleContexts.size(),
     _nrMaxContexts
   };
 }
@@ -1311,17 +1331,17 @@ bool V8DealerFeature::loadJavaScriptFileInContext(TRI_vocbase_t* vocbase,
     return false;
   }
 
-  enterContextInternal(vocbase, context, true);
+  context->lockAndEnter();
+  prepareLockedContext(vocbase, context, true);
+  TRI_DEFER(exitContextInternal(context));
 
   try {
     loadJavaScriptFileInternal(file, context, builder);
   } catch (...) {
     LOG_TOPIC(WARN, Logger::V8) << "caught exception while executing JavaScript file '" << file << "' in context #" << context->id();
-    exitContextInternal(context);
     throw;
   }
   
-  exitContextInternal(context);
   return true;
 }
 
diff --git a/arangod/V8Server/V8DealerFeature.h b/arangod/V8Server/V8DealerFeature.h
index 6fe003bb79..8624f2690a 100644
--- a/arangod/V8Server/V8DealerFeature.h
+++ b/arangod/V8Server/V8DealerFeature.h
@@ -131,19 +131,17 @@ class V8DealerFeature final : public application_features::ApplicationFeature {
   V8Context* buildContext(size_t id);
   V8Context* pickFreeContextForGc();
   void shutdownContext(V8Context* context);
-  void unblockContextsModification();
+  void unblockDynamicContextCreation();
   void loadJavaScriptFileInternal(std::string const& file, V8Context* context,
                                   VPackBuilder* builder);
   bool loadJavaScriptFileInContext(TRI_vocbase_t*, std::string const& file, V8Context* context, VPackBuilder* builder);
-  void enterContextInternal(TRI_vocbase_t* vocbase, V8Context* context, bool allowUseDatabase);
-  void enterLockedContext(TRI_vocbase_t*, V8Context*, bool allowUseDatabase);
+  void prepareLockedContext(TRI_vocbase_t*, V8Context*, bool allowUseDatabase);
   void exitContextInternal(V8Context*);
-  void exitLockedContext(V8Context*);
+  void cleanupLockedContext(V8Context*);
   void applyContextUpdate(V8Context* context);
   void shutdownContexts();
 
  private:
-  std::atomic<bool> _ok;
   std::atomic<uint64_t> _nextId;
 
   std::unique_ptr<Thread> _gcThread;
@@ -152,10 +150,10 @@ class V8DealerFeature final : public application_features::ApplicationFeature {
 
   basics::ConditionVariable _contextCondition;
   std::vector<V8Context*> _contexts;
-  std::vector<V8Context*> _freeContexts;
+  std::vector<V8Context*> _idleContexts;
   std::vector<V8Context*> _dirtyContexts;
   std::unordered_set<V8Context*> _busyContexts;
-  size_t _contextsModificationBlockers;
+  size_t _dynamicContextCreationBlockers;
 
   JSLoader _startupLoader;
 
diff --git a/arangod/V8Server/v8-actions.cpp b/arangod/V8Server/v8-actions.cpp
index c6bda9df20..5c0679bd03 100644
--- a/arangod/V8Server/v8-actions.cpp
+++ b/arangod/V8Server/v8-actions.cpp
@@ -1079,11 +1079,11 @@ static void JS_ExecuteGlobalContextFunction(
   // extract the action name
   v8::String::Utf8Value utf8def(args[0]);
 
-  if (*utf8def == 0) {
+  if (*utf8def == nullptr) {
     TRI_V8_THROW_TYPE_ERROR("<definition> must be a UTF-8 function definition");
   }
 
-  std::string const def = *utf8def;
+  std::string const def = std::string(*utf8def, utf8def.length());
 
   // and pass it to the V8 contexts
   if (!V8DealerFeature::DEALER->addGlobalContextMethod(def)) {
diff --git a/arangod/V8Server/v8-collection.cpp b/arangod/V8Server/v8-collection.cpp
index 194fe12351..009dddd0e7 100644
--- a/arangod/V8Server/v8-collection.cpp
+++ b/arangod/V8Server/v8-collection.cpp
@@ -957,7 +957,7 @@ static void JS_DropVocbaseCol(v8::FunctionCallbackInfo<v8::Value> const& args) {
       allowDropSystem = TRI_ObjectToBoolean(args[0]);
     }
   }
-  
+ 
   Result res = methods::Collections::drop(vocbase, collection,
                                           allowDropSystem, timeout);
   if (res.fail()) {
@@ -2396,14 +2396,14 @@ static void JS_StatusVocbaseCol(
       std::shared_ptr<LogicalCollection> const ci =
           ClusterInfo::instance()->getCollection(databaseName,
                                                  collection->cid_as_string());
-      TRI_V8_RETURN(v8::Number::New(isolate, (int)ci->getStatusLocked()));
+      TRI_V8_RETURN(v8::Number::New(isolate, (int)ci->status()));
     } catch (...) {
       TRI_V8_RETURN(v8::Number::New(isolate, (int)TRI_VOC_COL_STATUS_DELETED));
     }
   }
   // intentionally falls through
-
-  TRI_vocbase_col_status_e status = collection->getStatusLocked();
+  
+  TRI_vocbase_col_status_e status = collection->status();
 
   TRI_V8_RETURN(v8::Number::New(isolate, (int)status));
   TRI_V8_TRY_CATCH_END
diff --git a/arangod/V8Server/v8-vocbase.cpp b/arangod/V8Server/v8-vocbase.cpp
index f8135efab1..6718c48026 100644
--- a/arangod/V8Server/v8-vocbase.cpp
+++ b/arangod/V8Server/v8-vocbase.cpp
@@ -1387,7 +1387,12 @@ static void MapGetVocBase(v8::Local<v8::String> const name,
 
     // check if the collection is from the same database
     if (collection != nullptr && collection->vocbase() == vocbase) {
-      TRI_vocbase_col_status_e status = collection->getStatusLocked();
+      // we cannot use collection->getStatusLocked() here, because we
+      // have no idea who is calling us (db[...]). The problem is that
+      // if we are called from within a JavaScript transaction, the
+      // caller may have already acquired the collection's status lock
+      // with that transaction. if we now lock again, we may deadlock!
+      TRI_vocbase_col_status_e status = collection->status();
       TRI_voc_cid_t cid = collection->cid();
       uint32_t internalVersion = collection->internalVersion();
 
diff --git a/arangod/VocBase/AccessMode.h b/arangod/VocBase/AccessMode.h
index d921de35e2..01b6ffbc80 100644
--- a/arangod/VocBase/AccessMode.h
+++ b/arangod/VocBase/AccessMode.h
@@ -36,11 +36,18 @@ struct AccessMode {
     WRITE = 2,
     EXCLUSIVE = 4
   };
+  // no need to create an object of it
+  AccessMode() = delete; 
+
   static_assert(AccessMode::Type::NONE < AccessMode::Type::READ &&
                 AccessMode::Type::READ < AccessMode::Type::WRITE &&
                 AccessMode::Type::READ < AccessMode::Type::EXCLUSIVE,
                 "AccessMode::Type total order fail");
   
+  static inline bool isNone(Type type) {
+    return (type == Type::NONE);
+  }
+  
   static inline bool isRead(Type type) {
     return (type == Type::READ);
   }
diff --git a/arangod/VocBase/Methods/Collections.cpp b/arangod/VocBase/Methods/Collections.cpp
index 9262e1845d..cf0d6b6d4c 100644
--- a/arangod/VocBase/Methods/Collections.cpp
+++ b/arangod/VocBase/Methods/Collections.cpp
@@ -34,7 +34,7 @@
 #include "GeneralServer/AuthenticationFeature.h"
 #include "RestServer/DatabaseFeature.h"
 #include "StorageEngine/PhysicalCollection.h"
-#include "Transaction/StandaloneContext.h"
+#include "Transaction/V8Context.h"
 #include "Utils/ExecContext.h"
 #include "Utils/SingleCollectionTransaction.h"
 #include "V8/v8-conv.h"
@@ -78,8 +78,8 @@ void Collections::enumerate(
 }
 
 Result methods::Collections::lookup(TRI_vocbase_t* vocbase,
-                                  std::string const& name,
-                                  FuncCallback func) {
+                                    std::string const& name,
+                                    FuncCallback func) {
   if (name.empty()) {
     return Result(TRI_ERROR_ARANGO_COLLECTION_NOT_FOUND);
   } 
@@ -240,7 +240,7 @@ Result Collections::load(TRI_vocbase_t* vocbase, LogicalCollection* coll) {
 #endif
   }
 
-  auto ctx = transaction::StandaloneContext::Create(vocbase);
+  auto ctx = transaction::V8Context::CreateWhenRequired(vocbase, true);
   SingleCollectionTransaction trx(ctx, coll->cid(), AccessMode::Type::READ);
 
   Result res = trx.begin();
@@ -280,7 +280,7 @@ Result Collections::properties(LogicalCollection* coll, VPackBuilder& builder) {
 
   std::unique_ptr<SingleCollectionTransaction> trx;
   if (!ServerState::instance()->isCoordinator()) {
-    auto ctx = transaction::StandaloneContext::Create(coll->vocbase());
+    auto ctx = transaction::V8Context::CreateWhenRequired(coll->vocbase(), true);
     trx.reset(new SingleCollectionTransaction(ctx, coll->cid(),
                                               AccessMode::Type::READ));
     trx->addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
@@ -319,7 +319,7 @@ Result Collections::updateProperties(LogicalCollection* coll,
     auto info = ci->getCollection(coll->dbName(), coll->cid_as_string());
     return info->updateProperties(props, false);
   } else {
-    auto ctx = transaction::StandaloneContext::Create(coll->vocbase());
+    auto ctx = transaction::V8Context::CreateWhenRequired(coll->vocbase(), false);
     SingleCollectionTransaction trx(ctx, coll->cid(),
                                     AccessMode::Type::EXCLUSIVE);
     Result res = trx.begin();
@@ -484,7 +484,7 @@ Result Collections::warmup(TRI_vocbase_t* vocbase, LogicalCollection* coll) {
     return warmupOnCoordinator(vocbase->name(), cid);
   }
 
-  auto ctx = transaction::StandaloneContext::Create(vocbase);
+  auto ctx = transaction::V8Context::CreateWhenRequired(vocbase, false);
   SingleCollectionTransaction trx(ctx, coll->cid(), AccessMode::Type::READ);
   Result res = trx.begin();
   if (res.fail()) {
@@ -517,16 +517,19 @@ Result Collections::revisionId(TRI_vocbase_t* vocbase,
   
   if (ServerState::instance()->isCoordinator()) {
     return revisionOnCoordinator(databaseName, cid, rid);
-  } else {
-    auto ctx = transaction::StandaloneContext::Create(vocbase);
-    SingleCollectionTransaction trx(ctx, coll->cid(),
-                                    AccessMode::Type::READ);
-    Result res = trx.begin();
-    if (res.fail()) {
-      THROW_ARANGO_EXCEPTION(res);
-    }
-    rid = coll->revision(&trx);
-    return TRI_ERROR_NO_ERROR;
+  } 
+  
+  auto ctx = transaction::V8Context::CreateWhenRequired(vocbase, true);
+  SingleCollectionTransaction trx(ctx, coll->cid(),
+                                  AccessMode::Type::READ);
+  
+  Result res = trx.begin();
+  
+  if (res.fail()) {
+    THROW_ARANGO_EXCEPTION(res);
   }
+
+  rid = coll->revision(&trx);
+  return TRI_ERROR_NO_ERROR;
 }
 
diff --git a/arangod/VocBase/Methods/Indexes.cpp b/arangod/VocBase/Methods/Indexes.cpp
index 3a464a2404..af2af8f255 100644
--- a/arangod/VocBase/Methods/Indexes.cpp
+++ b/arangod/VocBase/Methods/Indexes.cpp
@@ -43,6 +43,7 @@
 #include "Transaction/Helpers.h"
 #include "Transaction/Hints.h"
 #include "Transaction/StandaloneContext.h"
+#include "Transaction/V8Context.h"
 #include "Utils/Events.h"
 #include "Utils/ExecContext.h"
 #include "Utils/SingleCollectionTransaction.h"
@@ -248,7 +249,7 @@ static Result EnsureIndexLocal(arangodb::LogicalCollection* collection,
   READ_LOCKER(readLocker, collection->vocbase()->_inventoryLock);
 
   SingleCollectionTransaction trx(
-      transaction::StandaloneContext::Create(collection->vocbase()),
+      transaction::V8Context::CreateWhenRequired(collection->vocbase(), false),
       collection->cid(),
       create ? AccessMode::Type::EXCLUSIVE : AccessMode::Type::READ);
 
@@ -260,17 +261,12 @@ static Result EnsureIndexLocal(arangodb::LogicalCollection* collection,
   bool created = false;
   std::shared_ptr<arangodb::Index> idx;
   if (create) {
-    // TODO Encapsulate in try{}catch(){} instead of errno()
     try {
       idx = collection->createIndex(&trx, definition, created);
     } catch (arangodb::basics::Exception const& e) {
-      return Result(e.code());
-    }
-    if (idx == nullptr) {
-      // something went wrong during creation
-      int res = TRI_errno();
-      return Result(res);
+      return Result(e.code(), e.what());
     }
+    TRI_ASSERT(idx != nullptr);
   } else {
     idx = collection->lookupIndex(definition);
     if (idx == nullptr) {
@@ -278,6 +274,8 @@ static Result EnsureIndexLocal(arangodb::LogicalCollection* collection,
       return Result(TRI_ERROR_ARANGO_INDEX_NOT_FOUND);
     }
   }
+    
+  TRI_ASSERT(idx != nullptr);
 
   VPackBuilder tmp;
   try {
@@ -541,7 +539,7 @@ arangodb::Result Indexes::drop(LogicalCollection const* collection,
     READ_LOCKER(readLocker, collection->vocbase()->_inventoryLock);
 
     SingleCollectionTransaction trx(
-        transaction::StandaloneContext::Create(collection->vocbase()),
+        transaction::V8Context::CreateWhenRequired(collection->vocbase(), false),
         collection->cid(), AccessMode::Type::EXCLUSIVE);
 
     Result res = trx.begin();
diff --git a/arangod/VocBase/vocbase.cpp b/arangod/VocBase/vocbase.cpp
index 76095c641c..00ad457141 100644
--- a/arangod/VocBase/vocbase.cpp
+++ b/arangod/VocBase/vocbase.cpp
@@ -1118,7 +1118,6 @@ int TRI_vocbase_t::dropCollection(arangodb::LogicalCollection* collection,
     int res;
     {
       READ_LOCKER(readLocker, _inventoryLock);
-
       res = dropCollectionWorker(collection, state, timeout);
     }
 
diff --git a/js/server/tests/replication/replication-sync.js b/js/server/tests/replication/replication-sync.js
index 393f0db4e2..3a2a0a945d 100644
--- a/js/server/tests/replication/replication-sync.js
+++ b/js/server/tests/replication/replication-sync.js
@@ -1260,7 +1260,6 @@ function ReplicationIncrementalKeyConflict() {
 
     setUp : function() {
       connectToMaster();
-
       db._drop(cn);
     },
 
@@ -1391,7 +1390,6 @@ function ReplicationIncrementalKeyConflict() {
 
 jsunity.run(ReplicationSuite);
 jsunity.run(ReplicationOtherDBSuite);
-// TODO: activate this test once it works
 jsunity.run(ReplicationIncrementalKeyConflict);
 
 return jsunity.done();
diff --git a/lib/ApplicationFeatures/ApplicationServer.cpp b/lib/ApplicationFeatures/ApplicationServer.cpp
index 240b8d4b31..2c4b27c0bc 100644
--- a/lib/ApplicationFeatures/ApplicationServer.cpp
+++ b/lib/ApplicationFeatures/ApplicationServer.cpp
@@ -49,7 +49,10 @@ ApplicationServer* ApplicationServer::server = nullptr;
 
 ApplicationServer::ApplicationServer(std::shared_ptr<ProgramOptions> options,
     const char *binaryPath)
-    : _options(options), _stopping(false), _binaryPath(binaryPath) {
+    : _state(ServerState::UNINITIALIZED),
+      _options(options),
+      _stopping(false),
+      _binaryPath(binaryPath) {
   // register callback function for failures
   fail = failCallback;
   
@@ -172,8 +175,8 @@ void ApplicationServer::run(int argc, char* argv[]) {
   
   // collect options from all features
   // in this phase, all features are order-independent
-  _state = ServerState::IN_COLLECT_OPTIONS;
-  reportServerProgress(_state);
+  _state.store(ServerState::IN_COLLECT_OPTIONS, std::memory_order_relaxed);
+  reportServerProgress(ServerState::IN_COLLECT_OPTIONS);
   collectOptions();
 
   // setup dependency, but ignore any failure for now
@@ -192,8 +195,8 @@ void ApplicationServer::run(int argc, char* argv[]) {
   _options->seal();
 
   // validate options of all features
-  _state = ServerState::IN_VALIDATE_OPTIONS;
-  reportServerProgress(_state);
+  _state.store(ServerState::IN_VALIDATE_OPTIONS, std::memory_order_relaxed);
+  reportServerProgress(ServerState::IN_VALIDATE_OPTIONS);
   validateOptions();
 
   // setup and validate all feature dependencies
@@ -211,8 +214,8 @@ void ApplicationServer::run(int argc, char* argv[]) {
   // furthermore, they must not write any files under elevated privileges
   // if they want other features to access them, or if they want to access
   // these files with dropped privileges
-  _state = ServerState::IN_PREPARE;
-  reportServerProgress(_state);
+  _state.store(ServerState::IN_PREPARE, std::memory_order_relaxed);
+  reportServerProgress(ServerState::IN_PREPARE);
   prepare();
   
   // turn off all features that depend on other features that have been
@@ -224,28 +227,28 @@ void ApplicationServer::run(int argc, char* argv[]) {
   dropPrivilegesPermanently();
 
   // start features. now features are allowed to start threads, write files etc.
-  _state = ServerState::IN_START;
-  reportServerProgress(_state);
+  _state.store(ServerState::IN_START, std::memory_order_relaxed);
+  reportServerProgress(ServerState::IN_START);
   start();
 
   // wait until we get signaled the shutdown request
-  _state = ServerState::IN_WAIT;
-  reportServerProgress(_state);
+  _state.store(ServerState::IN_WAIT, std::memory_order_relaxed);
+  reportServerProgress(ServerState::IN_WAIT);
   wait();
 
   // stop all features
-  _state = ServerState::IN_STOP;
-  reportServerProgress(_state);
+  _state.store(ServerState::IN_STOP, std::memory_order_relaxed);
+  reportServerProgress(ServerState::IN_STOP);
   stop();
 
   // unprepare all features
-  _state = ServerState::IN_UNPREPARE;
-  reportServerProgress(_state);
+  _state.store(ServerState::IN_UNPREPARE, std::memory_order_relaxed);
+  reportServerProgress(ServerState::IN_UNPREPARE);
   unprepare();
 
   // stopped
-  _state = ServerState::STOPPED;
-  reportServerProgress(_state);
+  _state.store(ServerState::STOPPED, std::memory_order_relaxed);
+  reportServerProgress(ServerState::STOPPED);
 }
 
 // signal the server to shut down
@@ -312,7 +315,7 @@ void ApplicationServer::collectOptions() {
       [this](ApplicationFeature* feature) {
         LOG_TOPIC(TRACE, Logger::STARTUP) << feature->name() << "::loadOptions";
         feature->collectOptions(_options);
-        reportFeatureProgress(_state, feature->name());
+        reportFeatureProgress(_state.load(std::memory_order_relaxed), feature->name());
       },
       true);
 }
@@ -369,7 +372,7 @@ void ApplicationServer::validateOptions() {
                                         << "::validateOptions";
       feature->validateOptions(_options);
       feature->state(FeatureState::VALIDATED);
-      reportFeatureProgress(_state, feature->name());
+      reportFeatureProgress(_state.load(std::memory_order_relaxed), feature->name());
     }
   }
 
@@ -561,7 +564,7 @@ void ApplicationServer::prepare() {
         throw;
       }
 
-      reportFeatureProgress(_state, feature->name());
+      reportFeatureProgress(_state.load(std::memory_order_relaxed), feature->name());
     }
   }
 }
@@ -581,7 +584,7 @@ void ApplicationServer::start() {
     try {
       feature->start();
       feature->state(FeatureState::STARTED);
-      reportFeatureProgress(_state, feature->name());
+      reportFeatureProgress(_state.load(std::memory_order_relaxed), feature->name());
     } catch (basics::Exception const& ex) {
       res.reset(ex.code(), std::string("startup aborted: caught exception during start of feature '") + feature->name() + "': " + ex.what());
     } catch (std::bad_alloc const& ex) {
@@ -660,7 +663,7 @@ void ApplicationServer::stop() {
       LOG_TOPIC(ERR, Logger::STARTUP) << "caught unknown exception during stop of feature '" << feature->name() << "'";
     }
     feature->state(FeatureState::STOPPED);
-    reportFeatureProgress(_state, feature->name());
+    reportFeatureProgress(_state.load(std::memory_order_relaxed), feature->name());
   }
 }
 
@@ -680,7 +683,7 @@ void ApplicationServer::unprepare() {
       LOG_TOPIC(ERR, Logger::STARTUP) << "caught unknown exception during unprepare of feature '" << feature->name() << "'";
     }
     feature->state(FeatureState::UNPREPARED);
-    reportFeatureProgress(_state, feature->name());
+    reportFeatureProgress(_state.load(std::memory_order_relaxed), feature->name());
   }
 }
 
diff --git a/lib/ApplicationFeatures/ApplicationServer.h b/lib/ApplicationFeatures/ApplicationServer.h
index 36315ab3d4..042a5b8450 100644
--- a/lib/ApplicationFeatures/ApplicationServer.h
+++ b/lib/ApplicationFeatures/ApplicationServer.h
@@ -134,9 +134,13 @@ class ApplicationServer {
   }
 
   static bool isPrepared() {
-    return server != nullptr && (server->_state == ServerState::IN_START ||
-                                 server->_state == ServerState::IN_WAIT ||
-                                 server->_state == ServerState::IN_STOP);
+    if (server != nullptr) {
+      ServerState tmp = server->_state.load(std::memory_order_relaxed);
+      return tmp == ServerState::IN_START ||
+             tmp == ServerState::IN_WAIT ||
+             tmp == ServerState::IN_STOP;
+    }
+    return false;
   }
 
   // returns the feature with the given name if known
@@ -291,7 +295,7 @@ class ApplicationServer {
 
  private:
   // the current state
-  ServerState _state = ServerState::UNINITIALIZED;
+  std::atomic<ServerState> _state;
 
   // the shared program options
   std::shared_ptr<options::ProgramOptions> _options;
diff --git a/lib/Basics/Mutex.cpp b/lib/Basics/Mutex.cpp
index 8018d7ca24..9ce969be66 100644
--- a/lib/Basics/Mutex.cpp
+++ b/lib/Basics/Mutex.cpp
@@ -35,7 +35,7 @@ using namespace arangodb;
 
 #if defined(TRI_HAVE_POSIX_THREADS)
 
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
 // initialize _holder to "maximum" thread id. this will work if the type of
 // _holder is numeric, but will not work if its type is more complex.
 Mutex::Mutex()
@@ -46,13 +46,13 @@ Mutex::Mutex() : _mutex() {
   pthread_mutexattr_init(&_attributes);
 
 #ifdef __linux__
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   // use an error checking mutex if available (only for LinuxThread) and only
   // in maintainer mode
   pthread_mutexattr_settype(&_attributes, PTHREAD_MUTEX_ERRORCHECK_NP);
 #endif
 #endif
-
+  
   pthread_mutex_init(&_mutex, &_attributes);
 }
 
@@ -62,7 +62,7 @@ Mutex::~Mutex() {
 }
 
 void Mutex::lock() {
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   // we must not hold the lock ourselves here
   TRI_ASSERT(_holder != Thread::currentThreadId());
 #endif
@@ -79,13 +79,13 @@ void Mutex::lock() {
     FATAL_ERROR_ABORT();
   }
 
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   _holder = Thread::currentThreadId();
 #endif
 }
 
 bool Mutex::tryLock() {
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   // we must not hold the lock ourselves here
   TRI_ASSERT(_holder != Thread::currentThreadId());
 #endif
@@ -104,7 +104,7 @@ bool Mutex::tryLock() {
     FATAL_ERROR_ABORT();
   }
 
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   _holder = Thread::currentThreadId();
 #endif
 
@@ -112,7 +112,7 @@ bool Mutex::tryLock() {
 }
 
 void Mutex::unlock() {
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   TRI_ASSERT(_holder == Thread::currentThreadId());
   _holder = 0;
 #endif
@@ -125,7 +125,7 @@ void Mutex::unlock() {
   }
 }
 
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
 void Mutex::assertLockedByCurrentThread() {
   TRI_ASSERT(_holder == Thread::currentThreadId());
 }
@@ -150,7 +150,7 @@ bool Mutex::tryLock() { return TryAcquireSRWLockExclusive(&_mutex) != 0; }
 
 void Mutex::unlock() { ReleaseSRWLockExclusive(&_mutex); }
 
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
 void Mutex::assertLockedByCurrentThread() {}
 void Mutex::assertNotLockedByCurrentThread() {}
 #endif
diff --git a/lib/Basics/Mutex.h b/lib/Basics/Mutex.h
index f9a12bf147..21dfa53a7a 100644
--- a/lib/Basics/Mutex.h
+++ b/lib/Basics/Mutex.h
@@ -28,6 +28,18 @@
 #include "Basics/Common.h"
 
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#define ARANGO_ENABLE_DEADLOCK_DETECTION
+#if defined(__SANITIZE_THREAD__)
+// Avoid fals positives with ThreadSanitizer
+#  undef ARANGO_ENABLE_DEADLOCK_DETECTION
+#elif defined(__has_feature)
+#  if __has_feature(thread_sanitizer)
+#    undef ARANGO_ENABLE_DEADLOCK_DETECTION
+#  endif
+#endif
+#endif
+
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
 #include "Basics/Thread.h"
 #endif
 
@@ -49,7 +61,7 @@ class Mutex {
   
   // assert that the mutex is locked by the current thread. will do
   // nothing in non-maintainer mode and will do nothing for non-posix locks
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   void assertLockedByCurrentThread();
   void assertNotLockedByCurrentThread();
 #else
@@ -68,7 +80,7 @@ class Mutex {
   SRWLOCK _mutex;
 #endif
 
-#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+#ifdef ARANGO_ENABLE_DEADLOCK_DETECTION
   TRI_tid_t _holder;
 #endif
 };