Sync Foxx Queues (#8254)

arangod/Cluster/HeartbeatThread.cpp

@@ -733,6 +733,7 @@ void HeartbeatThread::runSingleServer() {
         config._requireFromPresent = true;
         config._incremental = true;
         TRI_ASSERT(!config._skipCreateDrop);
+        config._includeFoxxQueues = true; // sync _queues and _jobs
 
         applier->forget();  // forget about any existing configuration
         applier->reconfigure(config);

arangod/MMFiles/MMFilesRestReplicationHandler.cpp

@@ -543,6 +543,7 @@ void MMFilesRestReplicationHandler::handleCommandInventory() {
 
   // include system collections?
   bool includeSystem = _request->parsedValue("includeSystem", true);
+  bool includeFoxxQueues = _request->parsedValue("includeFoxxQueues", false);
 
   // produce inventory for all databases?
   bool global = _request->parsedValue("global", false);
@@ -553,15 +554,15 @@ void MMFilesRestReplicationHandler::handleCommandInventory() {
         "global inventory can only be created from within _system database");
     return;
   }
-
+  
-  auto nameFilter = [includeSystem](LogicalCollection const* collection) {
+  auto nameFilter = [&](LogicalCollection const* collection) {
-    std::string const cname = collection->name();
+    std::string const& cname = collection->name();
-    if (!includeSystem && !cname.empty() && cname[0] == '_') {
+    if (!includeSystem && TRI_vocbase_t::IsSystemName(cname)) {
       // exclude all system collections
       return false;
     }
 
-    if (TRI_ExcludeCollectionReplication(cname, includeSystem)) {
+    if (TRI_ExcludeCollectionReplication(cname, includeSystem, includeFoxxQueues)) {
       // collection is excluded from replication
       return false;
     }

arangod/MMFiles/MMFilesWalAccess.cpp

@@ -253,7 +253,9 @@ struct MMFilesWalAccessContext : WalAccessContext {
         // will not find anything for a view
         LogicalCollection* collection = loadCollection(databaseId, datasourceId);
         if (collection != nullptr) {  // db may be already dropped
-          if (TRI_ExcludeCollectionReplication(collection->name(), _filter.includeSystem)) {
+          if (TRI_ExcludeCollectionReplication(collection->name(),
+                                               _filter.includeSystem,
+                                               _filter.includeFoxxQueues)) {
             return false;
           }
         }

arangod/MMFiles/mmfiles-replication-dump.cpp

@@ -323,7 +323,7 @@ static bool MustReplicateWalMarker(MMFilesReplicationDumpContext* dump,
   if (cid != 0) {
     std::string const& name = nameFromCid(dump, cid);
 
-    if (!name.empty() && TRI_ExcludeCollectionReplication(name, dump->_includeSystem)) {
+    if (!name.empty() && TRI_ExcludeCollectionReplication(name, dump->_includeSystem, /*includeFoxxQueues*/false)) {
       return false;
     }
   }

arangod/Replication/DatabaseInitialSyncer.cpp

@@ -1312,6 +1312,9 @@ arangodb::Result DatabaseInitialSyncer::fetchInventory(VPackBuilder& builder) {
   if (_config.applier._includeSystem) {
     url += "&includeSystem=true";
   }
+  if (_config.applier._includeFoxxQueues) {
+    url += "&includeFoxxQueues=true";
+  }
 
   // send request
   _config.progress.set("fetching master inventory from " + url);
@@ -1385,7 +1388,8 @@ Result DatabaseInitialSyncer::handleCollectionsAndViews(VPackSlice const& collSl
                     "collection name is missing in response");
     }
 
-    if (TRI_ExcludeCollectionReplication(masterName, _config.applier._includeSystem)) {
+    if (TRI_ExcludeCollectionReplication(masterName, _config.applier._includeSystem,
+                                         _config.applier._includeFoxxQueues)) {
       continue;
     }
 

arangod/Replication/GlobalInitialSyncer.cpp

@@ -371,6 +371,9 @@ Result GlobalInitialSyncer::fetchInventory(VPackBuilder& builder) {
   if (_state.applier._includeSystem) {
     url += "&includeSystem=true";
   }
+  if (_state.applier._includeFoxxQueues) {
+    url += "&includeFoxxQueues=true";
+  }
 
   // send request
   std::unique_ptr<httpclient::SimpleHttpResult> response;

arangod/Replication/ReplicationApplierConfiguration.cpp

@@ -63,7 +63,8 @@ ReplicationApplierConfiguration::ReplicationApplierConfiguration()
       _incremental(false),
       _verbose(false),
       _restrictType(RestrictType::None),
-      _restrictCollections() {}
+      _restrictCollections(),
+      _includeFoxxQueues(false) {}
 
 /// @brief reset the configuration to defaults
 void ReplicationApplierConfiguration::reset() {
@@ -95,6 +96,7 @@ void ReplicationApplierConfiguration::reset() {
   _verbose = false;
   _restrictType = RestrictType::None;
   _restrictCollections.clear();
+  _includeFoxxQueues = false;
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
   _force32mode = false;
 #endif

arangod/Replication/ReplicationApplierConfiguration.h

@@ -65,6 +65,7 @@ class ReplicationApplierConfiguration {
   bool _verbose;
   RestrictType _restrictType;
   std::set<std::string> _restrictCollections;
+  bool _includeFoxxQueues; /// sync the _jobs collection
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
   bool _force32mode = false;  // force client to act like 3.2
 #endif

arangod/Replication/TailingSyncer.cpp

@@ -241,7 +241,8 @@ bool TailingSyncer::isExcludedCollection(std::string const& masterName) const {
     return true;
   }
 
-  if (TRI_ExcludeCollectionReplication(masterName, true)) {
+  if (TRI_ExcludeCollectionReplication(masterName, /*includeSystem*/true,
+                                       _state.applier._includeFoxxQueues)) {
     return true;
   }
 
@@ -1645,8 +1646,9 @@ void TailingSyncer::fetchMasterLog(std::shared_ptr<Syncer::JobSynchronizer> shar
         (firstRegularTick > fetchTick ? "&firstRegular=" + StringUtils::itoa(firstRegularTick)
                                       : "") +
         "&serverId=" + _state.localServerIdString +
-        "&includeSystem=" + (_state.applier._includeSystem ? "true" : "false");
+        "&includeSystem=" + (_state.applier._includeSystem ? "true" : "false") +
-
+        "&includeFoxxQueues=" + (_state.applier._includeFoxxQueues ? "true" : "false");
+    
     // send request
     setProgress(std::string("fetching master log from tick ") + StringUtils::itoa(fetchTick) +
                 ", last scanned tick " + StringUtils::itoa(lastScannedTick) +

arangod/Replication/common-defines.cpp

@@ -44,7 +44,8 @@ void TRI_GetTimeStampReplication(double timeStamp, char* dst, size_t maxLength)
   strftime(dst, maxLength, "%Y-%m-%dT%H:%M:%SZ", &tb);
 }
 
-bool TRI_ExcludeCollectionReplication(std::string const& name, bool includeSystem) {
+bool TRI_ExcludeCollectionReplication(std::string const& name, bool includeSystem,
+                                      bool includeFoxxQueues) {
   if (name.empty()) {
     // should not happen...
     return true;
@@ -63,10 +64,11 @@ bool TRI_ExcludeCollectionReplication(std::string const& name, bool includeSyste
   if (TRI_IsPrefixString(name.c_str(), "_statistics") ||
       name == "_configuration" || name == "_frontend" ||
       name == "_cluster_kickstarter_plans" || name == "_routing" ||
-      name == "_fishbowl" || name == "_foxxlog" || name == "_jobs" ||
+      name == "_fishbowl" || name == "_foxxlog" || name == "_sessions") {
-      name == "_queues" || name == "_sessions") {
     // these system collections will always be excluded
     return true;
+  } else if (!includeFoxxQueues && (name == "_jobs" || name == "_queues")) {
+    return true;
   }
 
   return false;

arangod/Replication/common-defines.h

@@ -69,7 +69,9 @@ void TRI_GetTimeStampReplication(char*, size_t);
 void TRI_GetTimeStampReplication(double, char*, size_t);
 
 /// @brief determine whether a collection should be included in replication
-bool TRI_ExcludeCollectionReplication(std::string const& name, bool includeSystem);
+bool TRI_ExcludeCollectionReplication(std::string const& name,
+                                      bool includeSystem,
+                                      bool includeFoxxQueues);
 
 }  // namespace arangodb
 

arangod/RestHandler/RestWalAccessHandler.cpp

@@ -109,6 +109,7 @@ bool RestWalAccessHandler::parseFilter(WalAccess::Filter& filter) {
   }
 
   filter.includeSystem = _request->parsedValue("includeSystem", filter.includeSystem);
+  filter.includeFoxxQueues = _request->parsedValue("includeFoxxQueues", false);
 
   // grab list of transactions from the body value
   if (_request->requestType() == arangodb::rest::RequestType::PUT) {

arangod/RocksDBEngine/RocksDBReplicationContext.cpp

@@ -30,6 +30,7 @@
 #include "Basics/VPackStringBufferAdapter.h"
 #include "Logger/Logger.h"
 #include "Replication/common-defines.h"
+#include "Replication/ReplicationFeature.h"
 #include "Replication/utilities.h"
 #include "RestServer/DatabaseFeature.h"
 #include "RocksDBEngine/RocksDBCollection.h"
@@ -192,20 +193,21 @@ std::tuple<Result, TRI_voc_cid_t, uint64_t> RocksDBReplicationContext::bindColle
 
 // returns inventory
 Result RocksDBReplicationContext::getInventory(TRI_vocbase_t& vocbase, bool includeSystem,
+                                               bool includeFoxxQueues,
                                                bool global, VPackBuilder& result) {
   {
     MUTEX_LOCKER(locker, _contextLock);
     lazyCreateSnapshot();
   }
 
-  auto nameFilter = [includeSystem](LogicalCollection const* collection) {
+  auto nameFilter = [&](LogicalCollection const* collection) {
-    std::string const cname = collection->name();
+    std::string const& cname = collection->name();
-    if (!includeSystem && !cname.empty() && cname[0] == '_') {
+    if (!includeSystem && TRI_vocbase_t::IsSystemName(cname)) {
       // exclude all system collections
       return false;
     }
 
-    if (TRI_ExcludeCollectionReplication(cname, includeSystem)) {
+    if (TRI_ExcludeCollectionReplication(cname, includeSystem, includeFoxxQueues)) {
       // collection is excluded from replication
       return false;
     }

arangod/RocksDBEngine/RocksDBReplicationContext.h

@@ -131,7 +131,8 @@ class RocksDBReplicationContext {
       TRI_vocbase_t& vocbase, std::string const& cname);
 
   // returns inventory
-  Result getInventory(TRI_vocbase_t& vocbase, bool includeSystem, bool global,
+  Result getInventory(TRI_vocbase_t& vocbase, bool includeSystem,
+                      bool includeFoxxQueues, bool global,
                       velocypack::Builder&);
 
   // ========================= Dump API =============================

arangod/RocksDBEngine/RocksDBReplicationTailing.cpp

@@ -563,7 +563,8 @@ class WALParser final : public rocksdb::WriteBatch::Handler {
       if (collection == nullptr) {
         return false;
       }
-      return !TRI_ExcludeCollectionReplication(collection->name(), _includeSystem);
+      return !TRI_ExcludeCollectionReplication(collection->name(), _includeSystem,
+                                               /*includeFoxxQueues*/ false);
     }
     return false;
   }

arangod/RocksDBEngine/RocksDBRestReplicationHandler.cpp

@@ -363,6 +363,7 @@ void RocksDBRestReplicationHandler::handleCommandInventory() {
   TRI_voc_tick_t tick = TRI_CurrentTickServer();
   // include system collections?
   bool includeSystem = _request->parsedValue("includeSystem", true);
+  bool includeFoxxQs = _request->parsedValue("includeFoxxQueues", false);
 
   // produce inventory for all databases?
   bool isGlobal = false;
@@ -375,10 +376,10 @@ void RocksDBRestReplicationHandler::handleCommandInventory() {
   Result res;
   if (isGlobal) {
     builder.add(VPackValue("databases"));
-    res = ctx->getInventory(_vocbase, includeSystem, true, builder);
+    res = ctx->getInventory(_vocbase, includeSystem, includeFoxxQs, true, builder);
   } else {
     grantTemporaryRights();
-    res = ctx->getInventory(_vocbase, includeSystem, false, builder);
+    res = ctx->getInventory(_vocbase, includeSystem, includeFoxxQs, false, builder);
     TRI_ASSERT(builder.hasKey("collections") && builder.hasKey("views"));
   }
 

arangod/StorageEngine/WalAccess.cpp

@@ -57,7 +57,9 @@ bool WalAccessContext::shouldHandleCollection(TRI_voc_tick_t dbid, TRI_voc_cid_t
     if (collection == nullptr) {
       return false;
     }
-    return !TRI_ExcludeCollectionReplication(collection->name(), _filter.includeSystem);
+    return !TRI_ExcludeCollectionReplication(collection->name(),
+                                             _filter.includeSystem,
+                                             _filter.includeFoxxQueues);
   }
   return false;
 }

arangod/StorageEngine/WalAccess.h

@@ -97,6 +97,9 @@ class WalAccess {
 
     /// In case collection is == 0,
     bool includeSystem = false;
+    
+    /// export _queues and _jobs collection
+    bool includeFoxxQueues = false;
 
     /// only output markers from this database
     TRI_voc_tick_t vocbase = 0;

js/server/modules/@arangodb/foxx/queues/manager.js

@@ -161,17 +161,18 @@ exports.manage = function () {
   }
 
   if (global.ArangoServerState.getFoxxmasterQueueupdate()) {
-    if (isCluster) {
+    if (!isCluster) {
-      // Reset jobs before updating the queue delay. Don't continue on errors,
-      // but retry later.
-      resetDeadJobsOnFirstRun();
-      var foxxQueues = require('@arangodb/foxx/queues');
-      foxxQueues._updateQueueDelay();
-    } else {
       // On a Foxxmaster change FoxxmasterQueueupdate is set to true
       // we use this to signify a Leader change to this server
       foxxManager.healAll(true);
     }
+    // Reset jobs before updating the queue delay. Don't continue on errors,
+    // but retry later.
+    resetDeadJobsOnFirstRun();
+    if (isCluster) {
+      var foxxQueues = require('@arangodb/foxx/queues');
+      foxxQueues._updateQueueDelay();
+    } 
     // do not call again immediately
     global.ArangoServerState.setFoxxmasterQueueupdate(false);
   }