text: ////////////////////////////////////////////////////////////////////////////////
  /// DISCLAIMER
  ///
  /// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
  /// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
  ///
  /// Licensed under the Apache License, Version 2.0 (the "License");
  /// you may not use this file except in compliance with the License.
  /// You may obtain a copy of the License at
  ///
  ///     http://www.apache.org/licenses/LICENSE-2.0
  ///
  /// Unless required by applicable law or agreed to in writing, software
  /// distributed under the License is distributed on an "AS IS" BASIS,
  /// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  /// See the License for the specific language governing permissions and
  /// limitations under the License.
  ///
  /// Copyright holder is ArangoDB GmbH, Cologne, Germany
  ///
  /// @author Kaveh Vahedipour
  /// @author Matthew Von-Maszewski
  ////////////////////////////////////////////////////////////////////////////////
  
  #include "SynchronizeShard.h"
  
  #include "Agency/AgencyStrings.h"
  #include "Agency/TimeString.h"
  #include "ApplicationFeatures/ApplicationServer.h"
  #include "Basics/VelocyPackHelper.h"
  #include "Cluster/ActionDescription.h"
  #include "Cluster/ClusterComm.h"
  #include "Cluster/ClusterFeature.h"
  #include "Cluster/FollowerInfo.h"
  #include "Cluster/MaintenanceFeature.h"
  #include "Cluster/ServerState.h"
  #include "Replication/DatabaseInitialSyncer.h"
  #include "Replication/DatabaseReplicationApplier.h"
  #include "Replication/DatabaseTailingSyncer.h"
  #include "Replication/GlobalInitialSyncer.h"
  #include "Replication/GlobalReplicationApplier.h"
  #include "Replication/ReplicationApplierConfiguration.h"
  #include "Replication/ReplicationFeature.h"
  #include "RestServer/DatabaseFeature.h"
  #include "RestServer/ServerIdFeature.h"
  #include "Transaction/StandaloneContext.h"
  #include "Utils/CollectionNameResolver.h"
  #include "Utils/SingleCollectionTransaction.h"
  #include "VocBase/LogicalCollection.h"
  #include "VocBase/Methods/Collections.h"
  #include "VocBase/Methods/Databases.h"
  
  #include <velocypack/Compare.h>
  #include <velocypack/Iterator.h>
  #include <velocypack/Slice.h>
  #include <velocypack/velocypack-aliases.h>
  
  using namespace arangodb::application_features;
  using namespace arangodb::maintenance;
  using namespace arangodb::methods;
  using namespace arangodb::transaction;
  using namespace arangodb;
  using namespace arangodb::consensus;
  
  std::string const ENDPOINT("endpoint");
  std::string const INCLUDE("include");
  std::string const INCLUDE_SYSTEM("includeSystem");
  std::string const INCREMENTAL("incremental");
  std::string const KEEP_BARRIER("keepBarrier");
  std::string const LEADER_ID("leaderId");
  std::string const BARRIER_ID("barrierId");
  std::string const LAST_LOG_TICK("lastLogTick");
  std::string const API_REPLICATION("/_api/replication/");
  std::string const REPL_ADD_FOLLOWER(API_REPLICATION + "addFollower");
  std::string const REPL_BARRIER_API(API_REPLICATION + "barrier/");
  std::string const REPL_HOLD_READ_LOCK(API_REPLICATION +
                                        "holdReadLockCollection");
  std::string const REPL_REM_FOLLOWER(API_REPLICATION + "removeFollower");
  std::string const RESTRICT_TYPE("restrictType");
  std::string const RESTRICT_COLLECTIONS("restrictCollections");
  std::string const SKIP_CREATE_DROP("skipCreateDrop");
  std::string const TTL("ttl");
  
  using namespace std::chrono;
  
  SynchronizeShard::SynchronizeShard(MaintenanceFeature& feature, ActionDescription const& desc)
      : ActionBase(feature, desc) {
    std::stringstream error;
  
    if (!desc.has(COLLECTION)) {
      error << "collection must be specified";
    }
    TRI_ASSERT(desc.has(COLLECTION));
  
    if (!desc.has(DATABASE)) {
      error << "database must be specified";
    }
    TRI_ASSERT(desc.has(DATABASE));
  
    if (!desc.has(SHARD)) {
      error << "SHARD must be specified";
    }
    TRI_ASSERT(desc.has(SHARD));
  
    if (!desc.has(THE_LEADER) || desc.get(THE_LEADER).empty()) {
      error << "leader must be specified and must be non-empty";
    }
    TRI_ASSERT(desc.has(THE_LEADER) && !desc.get(THE_LEADER).empty());
  
    if (!desc.has(SHARD_VERSION)) {
      error << "local shard version must be specified. ";
    }
    TRI_ASSERT(desc.has(SHARD_VERSION));
  
    if (!error.str().empty()) {
      LOG_TOPIC("03780", ERR, Logger::MAINTENANCE) << "SynchronizeShard: " << error.str();
      _result.reset(TRI_ERROR_INTERNAL, error.str());
      setState(FAILED);
    }
  }
  
  SynchronizeShard::~SynchronizeShard() {}
  
  class SynchronizeShardCallback : public arangodb::ClusterCommCallback {
   public:
    explicit SynchronizeShardCallback(SynchronizeShard* callee) {}
    virtual bool operator()(arangodb::ClusterCommResult*) override final {
      return true;
    }
  };
  
  static std::stringstream& AppendShardInformationToMessage(
      std::string const& database, std::string const& shard, std::string const& planId,
      std::chrono::system_clock::time_point const& startTime, std::stringstream& msg) {
    auto const endTime = system_clock::now();
    msg << "local shard: '" << database << "/" << shard << "', "
        << "for central: '" << database << "/" << planId << "', "
        << "started: " << timepointToString(startTime) << ", "
        << "ended: " << timepointToString(endTime);
    return msg;
  }
  
  static arangodb::Result getReadLockId(std::string const& endpoint,
                                        std::string const& database, std::string const& clientId,
                                        double timeout, uint64_t& id) {
    std::string error("startReadLockOnLeader: Failed to get read lock - ");
  
    auto cc = arangodb::ClusterComm::instance();
    if (cc == nullptr) {  // nullptr only happens during controlled shutdown
      return arangodb::Result(TRI_ERROR_SHUTTING_DOWN,
                              "startReadLockOnLeader: Shutting down");
    }
  
    auto comres =
        cc->syncRequest(TRI_NewTickServer(), endpoint, rest::RequestType::GET,
                        DB + database + REPL_HOLD_READ_LOCK, std::string(),
                        std::unordered_map<std::string, std::string>(), timeout);
  
    auto result = comres->result;
  
    if (result != nullptr && result->getHttpReturnCode() == 200) {
      auto const idv = result->getBodyVelocyPack();
      auto const& idSlice = idv->slice();
      TRI_ASSERT(idSlice.isObject());
      TRI_ASSERT(idSlice.hasKey(ID));
      try {
        id = std::stoull(idSlice.get(ID).copyString());
      } catch (std::exception const&) {
        error += " expecting id to be uint64_t ";
        error += idSlice.toJson();
        return arangodb::Result(TRI_ERROR_INTERNAL, error);
      }
    } else {
      if (result) {
        error.append(result->getHttpReturnMessage());
      } else {
        error.append(comres->stringifyErrorMessage());
      }
      return arangodb::Result(TRI_ERROR_INTERNAL, error);
    }
  
    return arangodb::Result();
  }
  
  static arangodb::Result collectionCount(std::shared_ptr<arangodb::LogicalCollection> const& col,
                                          uint64_t& c) {
    std::string collectionName(col->name());
    auto ctx = std::make_shared<transaction::StandaloneContext>(col->vocbase());
    SingleCollectionTransaction trx(ctx, collectionName, AccessMode::Type::READ);
  
    Result res = trx.begin();
    if (!res.ok()) {
      LOG_TOPIC("5be16", ERR, Logger::MAINTENANCE) << "Failed to start count transaction: " << res;
      return res;
    }
  
    OperationResult opResult =
        trx.count(collectionName, arangodb::transaction::CountType::Normal);
    res = trx.finish(opResult.result);
  
    if (res.fail()) {
      LOG_TOPIC("263d2", ERR, Logger::MAINTENANCE)
          << "Failed to finish count transaction: " << res;
      return res;
    }
  
    VPackSlice s = opResult.slice();
    TRI_ASSERT(s.isNumber());
    c = s.getNumber<uint64_t>();
  
    return opResult.result;
  }
  
  static arangodb::Result addShardFollower(std::string const& endpoint,
                                           std::string const& database,
                                           std::string const& shard, uint64_t lockJobId,
                                           std::string const& clientId,
                                           SyncerId const syncerId,
                                           std::string const& clientInfoString,
                                           double timeout = 120.0) {
    LOG_TOPIC("b982e", DEBUG, Logger::MAINTENANCE)
        << "addShardFollower: tell the leader to put us into the follower "
           "list...";
  
    auto cc = arangodb::ClusterComm::instance();
    if (cc == nullptr) {  // nullptr only happens during controlled shutdown
      return arangodb::Result(TRI_ERROR_SHUTTING_DOWN,
                              "startReadLockOnLeader: Shutting down");
    }
  
    try {
      DatabaseGuard guard(database);
      auto vocbase = &guard.database();
  
      auto collection = vocbase->lookupCollection(shard);
      if (collection == nullptr) {
        std::string errorMsg(
            "SynchronizeShard::addShardFollower: Failed to lookup collection ");
        errorMsg += shard;
        LOG_TOPIC("4a8db", ERR, Logger::MAINTENANCE) << errorMsg;
        return arangodb::Result(TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND, errorMsg);
      }
  
      uint64_t docCount;
      Result res = collectionCount(collection, docCount);
      if (res.fail()) {
        return res;
      }
      VPackBuilder body;
      {
        VPackObjectBuilder b(&body);
        body.add(FOLLOWER_ID, VPackValue(arangodb::ServerState::instance()->getId()));
        body.add(SHARD, VPackValue(shard));
        body.add("checksum", VPackValue(std::to_string(docCount)));
        body.add("serverId", VPackValue(basics::StringUtils::itoa(ServerIdFeature::getId())));
        if (syncerId.value != 0) {
          body.add("syncerId", VPackValue(syncerId.toString()));
        }
        if (!clientInfoString.empty()) {
          body.add("clientInfo", VPackValue(clientInfoString));
        }
        if (lockJobId != 0) {
          body.add("readLockId", VPackValue(std::to_string(lockJobId)));
        } else {  // short cut case
          if (docCount != 0) {
            // This can happen if we once were an in-sync follower and a
            // synchronization request has timed out, but still runs on our
            // side here. In this case, we can simply continue with the slow
            // path and run the full sync protocol. Therefore we error out
            // here. Note that we are in the lockJobId == 0 case, which is
            // the shortcut.
            std::string msg =
                "Short cut synchronization for shard " + shard +
                " did not work, since we got a document in the meantime.";
            LOG_TOPIC("ef299", INFO, Logger::MAINTENANCE) << msg;
            return arangodb::Result(TRI_ERROR_INTERNAL, msg);
          }
        }
      }
  
      auto comres =
          cc->syncRequest(TRI_NewTickServer(), endpoint, rest::RequestType::PUT,
                          DB + database + REPL_ADD_FOLLOWER, body.toJson(),
                          std::unordered_map<std::string, std::string>(), timeout);
  
      auto result = comres->result;
      std::string errorMessage(
          "addShardFollower: could not add us to the leader's follower list. ");
      if (result == nullptr || result->getHttpReturnCode() != 200) {
        if (lockJobId != 0) {
          errorMessage += comres->stringifyErrorMessage();
          LOG_TOPIC("22e0a", ERR, Logger::MAINTENANCE) << errorMessage;
        } else {
          errorMessage += "With shortcut (can happen, no problem).";
          LOG_TOPIC("abf2e", INFO, Logger::MAINTENANCE) << errorMessage;
        }
        return arangodb::Result(TRI_ERROR_INTERNAL, errorMessage);
      }
  
      LOG_TOPIC("79935", DEBUG, Logger::MAINTENANCE) << "addShardFollower: success";
      return arangodb::Result();
    } catch (std::exception const& e) {
      std::string errorMsg(
          "SynchronizeShard::addShardFollower: Failed to lookup database ");
      errorMsg += database;
      errorMsg += " exception: ";
      errorMsg += e.what();
      LOG_TOPIC("6b7e8", ERR, Logger::MAINTENANCE) << errorMsg;
      return arangodb::Result(TRI_ERROR_ARANGO_DATABASE_NOT_FOUND, errorMsg);
    }
  }
  
  static arangodb::Result cancelReadLockOnLeader(std::string const& endpoint,
                                                 std::string const& database, uint64_t lockJobId,
                                                 std::string const& clientId,
                                                 double timeout = 10.0) {
    auto cc = arangodb::ClusterComm::instance();
    if (cc == nullptr) {  // nullptr only happens during controlled shutdown
      return arangodb::Result(TRI_ERROR_SHUTTING_DOWN,
                              "cancelReadLockOnLeader: Shutting down");
    }
  
    VPackBuilder body;
    {
      VPackObjectBuilder b(&body);
      body.add(ID, VPackValue(std::to_string(lockJobId)));
    }
  
    auto comres =
        cc->syncRequest(TRI_NewTickServer(), endpoint, rest::RequestType::DELETE_REQ,
                        DB + database + REPL_HOLD_READ_LOCK, body.toJson(),
                        std::unordered_map<std::string, std::string>(), timeout);
  
    auto result = comres->result;
  
    if (result != nullptr && result->getHttpReturnCode() == 404) {
      auto const vp = result->getBodyVelocyPack();
      auto const& slice = vp->slice();
      if (slice.isObject()) {
        VPackSlice s = slice.get(StaticStrings::ErrorNum);
        if (s.isNumber()) {
          int errorNum = s.getNumber<int>();
          if (errorNum == TRI_ERROR_ARANGO_DATABASE_NOT_FOUND) {
            // database is gone. that means our lock is also gone
            return arangodb::Result();
          }
        }
      }
    }
  
    if (result == nullptr || result->getHttpReturnCode() != 200) {
      auto errorMessage = comres->stringifyErrorMessage();
      LOG_TOPIC("52924", ERR, Logger::MAINTENANCE)
          << "cancelReadLockOnLeader: exception caught for " << body.toJson()
          << ": " << errorMessage;
      return arangodb::Result(TRI_ERROR_INTERNAL, errorMessage);
    }
  
    LOG_TOPIC("4355c", DEBUG, Logger::MAINTENANCE) << "cancelReadLockOnLeader: success";
    return arangodb::Result();
  }
  
  static arangodb::Result cancelBarrier(std::string const& endpoint,
                                        std::string const& database, int64_t barrierId,
                                        std::string const& clientId, double timeout = 120.0) {
    if (barrierId <= 0) {
      return Result();
    }
  
    auto cc = arangodb::ClusterComm::instance();
    if (cc == nullptr) {  // nullptr only happens during controlled shutdown
      return arangodb::Result(TRI_ERROR_SHUTTING_DOWN,
                              "startReadLockOnLeader: Shutting down");
    }
  
    auto comres =
        cc->syncRequest(TRI_NewTickServer(), endpoint, rest::RequestType::DELETE_REQ,
                        DB + database + REPL_BARRIER_API + std::to_string(barrierId),
                        std::string(),
                        std::unordered_map<std::string, std::string>(), timeout);
  
    // I'm sure that syncRequest cannot return null. But the check doesn't hurt
    // and is preferable over a segfault.
    TRI_ASSERT(comres != nullptr);
    if (comres == nullptr) {
      LOG_TOPIC("00aa2", ERR, Logger::MAINTENANCE)
          << "CancelBarrier: error: syncRequest returned null";
      return arangodb::Result{TRI_ERROR_INTERNAL};
    }
  
    if (comres->status == CL_COMM_SENT) {
      auto result = comres->result;
      if (result == nullptr ||
          (result->getHttpReturnCode() != 200 && result->getHttpReturnCode() != 204)) {
        std::string errorMessage = comres->stringifyErrorMessage();
        LOG_TOPIC("f5733", ERR, Logger::MAINTENANCE) << "CancelBarrier: error" << errorMessage;
        return arangodb::Result(TRI_ERROR_INTERNAL, errorMessage);
      }
    } else {
      std::string error(
          "CancelBarrier: failed to send message to leader : status ");
      error += ClusterCommResult::stringifyStatus(comres->status);
      LOG_TOPIC("1c48a", ERR, Logger::MAINTENANCE) << error;
      return arangodb::Result(TRI_ERROR_INTERNAL, error);
    }
  
    LOG_TOPIC("313dc", DEBUG, Logger::MAINTENANCE) << "cancelBarrier: success";
    return arangodb::Result();
  }
  
  static inline bool isStopping() {
    return application_features::ApplicationServer::isStopping();
  }
  
  arangodb::Result SynchronizeShard::getReadLock(std::string const& endpoint,
                                                 std::string const& database,
                                                 std::string const& collection,
                                                 std::string const& clientId, uint64_t rlid,
                                                 bool soft, double timeout) {
    auto cc = arangodb::ClusterComm::instance();
    if (cc == nullptr) {  // nullptr only happens during controlled shutdown
      return arangodb::Result(TRI_ERROR_SHUTTING_DOWN,
                              "startReadLockOnLeader: Shutting down");
    }
  
    VPackBuilder body;
    {
      VPackObjectBuilder o(&body);
      body.add(ID, VPackValue(std::to_string(rlid)));
      body.add(COLLECTION, VPackValue(collection));
      body.add(TTL, VPackValue(timeout));
      body.add(StaticStrings::ReplicationSoftLockOnly, VPackValue(soft));
    }
  
    auto url = DB + database + REPL_HOLD_READ_LOCK;
  
    cc->asyncRequest(TRI_NewTickServer(), endpoint, rest::RequestType::POST, url,
                     std::make_shared<std::string>(body.toJson()),
                     std::unordered_map<std::string, std::string>(),
                     std::make_shared<SynchronizeShardCallback>(this), timeout, true, timeout);
  
    // Intentionally do not look at the outcome, even in case of an error
    // we must make sure that the read lock on the leader is not active!
    // This is done automatically below.
  
    double sleepTime = 0.5;
    size_t count = 0;
    size_t maxTries = static_cast<size_t>(std::floor(600.0 / sleepTime));
    while (++count < maxTries) {  // wait for some time until read lock established:
  
      if (isStopping()) {
        return arangodb::Result(TRI_ERROR_SHUTTING_DOWN);
      }
  
      // Now check that we hold the read lock:
      auto putres = cc->syncRequest(TRI_NewTickServer(), endpoint,
                                    rest::RequestType::PUT, url, body.toJson(),
                                    std::unordered_map<std::string, std::string>(), timeout);
  
      auto result = putres->result;
      if (result != nullptr && result->getHttpReturnCode() == 200) {
        auto const vp = putres->result->getBodyVelocyPack();
        auto const& slice = vp->slice();
        TRI_ASSERT(slice.isObject());
        VPackSlice lockHeld = slice.get("lockHeld");
        if (lockHeld.isBoolean() && lockHeld.getBool()) {
          return arangodb::Result();
        }
        LOG_TOPIC("b681f", DEBUG, Logger::MAINTENANCE)
            << "startReadLockOnLeader: Lock not yet acquired...";
      } else {
        if (result != nullptr && result->getHttpReturnCode() == 404) {
          auto const vp = result->getBodyVelocyPack();
          auto const& slice = vp->slice();
          if (slice.isObject()) {
            VPackSlice s = slice.get(StaticStrings::ErrorNum);
            if (s.isNumber()) {
              int errorNum = s.getNumber<int>();
              if (errorNum == TRI_ERROR_ARANGO_DATABASE_NOT_FOUND) {
                // database is gone. we can now give up
                break;
              }
            }
          }
          // fall-through to other cases intentional here
        }
  
        LOG_TOPIC("a82bc", DEBUG, Logger::MAINTENANCE)
            << "startReadLockOnLeader: Do not see read lock yet:"
            << putres->stringifyErrorMessage();
      }
  
      std::this_thread::sleep_for(duration<double>(sleepTime));
    }
  
    LOG_TOPIC("75e2b", ERR, Logger::MAINTENANCE) << "startReadLockOnLeader: giving up";
  
    try {
      auto r = cc->syncRequest(TRI_NewTickServer(), endpoint,
                               rest::RequestType::DELETE_REQ, url, body.toJson(),
                               std::unordered_map<std::string, std::string>(), timeout);
      if (r->result == nullptr || r->result->getHttpReturnCode() != 200) {
        LOG_TOPIC("4f34d", ERR, Logger::MAINTENANCE)
            << "startReadLockOnLeader: cancelation error for shard - " << collection
            << " " << r->getErrorCode() << ": " << r->stringifyErrorMessage();
      }
    } catch (std::exception const& e) {
      LOG_TOPIC("7fcc9", ERR, Logger::MAINTENANCE)
          << "startReadLockOnLeader: exception in cancel: " << e.what();
    }
  
    return arangodb::Result(TRI_ERROR_CLUSTER_TIMEOUT,
                            "startReadLockOnLeader: giving up");
  }
  
  arangodb::Result SynchronizeShard::startReadLockOnLeader(
      std::string const& endpoint, std::string const& database, std::string const& collection,
      std::string const& clientId, uint64_t& rlid, bool soft, double timeout) {
    // Read lock id
    rlid = 0;
    arangodb::Result result = getReadLockId(endpoint, database, clientId, timeout, rlid);
    if (!result.ok()) {
      LOG_TOPIC("2e5ae", ERR, Logger::MAINTENANCE) << result.errorMessage();
      return result;
    }
    LOG_TOPIC("c8d18", DEBUG, Logger::MAINTENANCE) << "Got read lock id: " << rlid;
  
    result = getReadLock(endpoint, database, collection, clientId, rlid, soft, timeout);
  
    return result;
  }
  
  enum ApplierType { APPLIER_DATABASE, APPLIER_GLOBAL };
  
  static arangodb::ResultT<SyncerId> replicationSynchronize(
      std::shared_ptr<arangodb::LogicalCollection> const& col,
      VPackSlice const& config, std::string const& clientInfoString,
      ApplierType applierType, std::shared_ptr<VPackBuilder> sy) {
    auto& vocbase = col->vocbase();
  
    auto database = vocbase.name();
  
    auto shard = col->name();
  
    bool keepBarrier = config.get(KEEP_BARRIER).getBool();
    std::string leaderId;
    if (config.hasKey(LEADER_ID)) {
      leaderId = config.get(LEADER_ID).copyString();
    }
  
    ReplicationApplierConfiguration configuration =
        ReplicationApplierConfiguration::fromVelocyPack(config, database);
    configuration.setClientInfo(clientInfoString);
    configuration.validate();
  
    std::shared_ptr<InitialSyncer> syncer;
  
    if (applierType == APPLIER_DATABASE) {
      // database-specific synchronization
      syncer.reset(new DatabaseInitialSyncer(vocbase, configuration));
  
      if (!leaderId.empty()) {
        syncer->setLeaderId(leaderId);
      }
    } else if (applierType == APPLIER_GLOBAL) {
      configuration._skipCreateDrop = false;
      syncer.reset(new GlobalInitialSyncer(configuration));
    } else {
      TRI_ASSERT(false);
    }
  
    SyncerId syncerId{syncer->syncerId()};
  
    try {
      Result r = syncer->run(configuration._incremental);
  
      if (r.fail()) {
        LOG_TOPIC("3efff", DEBUG, Logger::REPLICATION)
            << "initial sync failed for database '" << database
            << "': " << r.errorMessage();
        THROW_ARANGO_EXCEPTION_MESSAGE(r.errorNumber(), r.errorMessage());
      }
  
      {
        VPackObjectBuilder o(sy.get());
        if (keepBarrier) {
          sy->add(BARRIER_ID, VPackValue(syncer->stealBarrier()));
        }
        sy->add(LAST_LOG_TICK, VPackValue(syncer->getLastLogTick()));
        sy->add(VPackValue(COLLECTIONS));
        {
          VPackArrayBuilder a(sy.get());
          for (auto const& i : syncer->getProcessedCollections()) {
            VPackObjectBuilder e(sy.get());
            sy->add(ID, VPackValue(i.first));
            sy->add(NAME, VPackValue(i.second));
          }
        }
      }
  
    } catch (arangodb::basics::Exception const& ex) {
      std::string s("cannot sync from remote endpoint: ");
      s += ex.what() + std::string(". last progress message was '") +
           syncer->progress() + "'";
      return Result(ex.code(), s);
    } catch (std::exception const& ex) {
      std::string s("cannot sync from remote endpoint: ");
      s += ex.what() + std::string(". last progress message was '") +
           syncer->progress() + "'";
      return Result(TRI_ERROR_INTERNAL, s);
    } catch (...) {
      std::string s(
          "cannot sync from remote endpoint: unknown exception. last progress "
          "message was '");
      s += syncer->progress() + "'";
      return Result(TRI_ERROR_INTERNAL, s);
    }
  
    return ResultT<SyncerId>::success(syncerId);
  }
  
  static arangodb::Result replicationSynchronizeCatchup(VPackSlice const& conf, double timeout,
                                                        TRI_voc_tick_t& tickReached,
                                                        bool& didTimeout) {
    didTimeout = false;
  
    auto const database = conf.get(DATABASE).copyString();
    auto const collection = conf.get(COLLECTION).copyString();
    auto const leaderId = conf.get(LEADER_ID).copyString();
    auto const fromTick = conf.get("from").getNumber<uint64_t>();
  
    ReplicationApplierConfiguration configuration =
        ReplicationApplierConfiguration::fromVelocyPack(conf, database);
    // will throw if invalid
    configuration.validate();
  
    DatabaseGuard guard(database);
    DatabaseTailingSyncer syncer(guard.database(), configuration, fromTick, true, 0);
  
    if (!leaderId.empty()) {
      syncer.setLeaderId(leaderId);
    }
  
    Result r;
    try {
      r = syncer.syncCollectionCatchup(collection, timeout, tickReached, didTimeout);
    } catch (arangodb::basics::Exception const& ex) {
      r = Result(ex.code(), ex.what());
    } catch (std::exception const& ex) {
      r = Result(TRI_ERROR_INTERNAL, ex.what());
    } catch (...) {
      r = Result(TRI_ERROR_INTERNAL, "unknown exception");
    }
  
    if (r.fail()) {
      LOG_TOPIC("fa2ab", ERR, Logger::REPLICATION)
          << "syncCollectionFinalize failed: " << r.errorMessage();
    }
  
    return r;
  }
  
  static arangodb::Result replicationSynchronizeFinalize(VPackSlice const& conf) {
    auto const database = conf.get(DATABASE).copyString();
    auto const collection = conf.get(COLLECTION).copyString();
    auto const leaderId = conf.get(LEADER_ID).copyString();
    auto const fromTick = conf.get("from").getNumber<uint64_t>();
  
    ReplicationApplierConfiguration configuration =
        ReplicationApplierConfiguration::fromVelocyPack(conf, database);
    // will throw if invalid
    configuration.validate();
  
    DatabaseGuard guard(database);
    DatabaseTailingSyncer syncer(guard.database(), configuration, fromTick, true, 0);
  
    if (!leaderId.empty()) {
      syncer.setLeaderId(leaderId);
    }
  
    Result r;
    try {
      r = syncer.syncCollectionFinalize(collection);
    } catch (arangodb::basics::Exception const& ex) {
      r = Result(ex.code(), ex.what());
    } catch (std::exception const& ex) {
      r = Result(TRI_ERROR_INTERNAL, ex.what());
    } catch (...) {
      r = Result(TRI_ERROR_INTERNAL, "unknown exception");
    }
  
    if (r.fail()) {
      LOG_TOPIC("e8056", ERR, Logger::REPLICATION)
          << "syncCollectionFinalize failed: " << r.errorMessage();
    }
  
    return r;
  }
  
  bool SynchronizeShard::first() {
    std::string database = _description.get(DATABASE);
    std::string planId = _description.get(COLLECTION);
    std::string shard = _description.get(SHARD);
    std::string leader = _description.get(THE_LEADER);
  
    LOG_TOPIC("fa651", DEBUG, Logger::MAINTENANCE)
        << "SynchronizeShard: synchronizing shard '" << database << "/" << shard
        << "' for central '" << database << "/" << planId << "'";
  
    auto* clusterInfo = ClusterInfo::instance();
    auto const ourselves = arangodb::ServerState::instance()->getId();
    auto startTime = system_clock::now();
    auto const startTimeStr = timepointToString(startTime);
    std::string const clientId(database + planId + shard + leader);
  
    // First wait until the leader has created the shard (visible in
    // Current in the Agency) or we or the shard have vanished from
    // the plan:
    while (true) {
      if (isStopping()) {
        _result.reset(TRI_ERROR_SHUTTING_DOWN);
        return false;
      }
  
      std::vector<std::string> planned;
      auto result = clusterInfo->getShardServers(shard, planned);
  
      if (!result.ok() ||
          std::find(planned.begin(), planned.end(), ourselves) == planned.end() ||
          planned.front() != leader) {
        // Things have changed again, simply terminate:
        std::stringstream error;
        error << "cancelled, ";
        AppendShardInformationToMessage(database, shard, planId, startTime, error);
        LOG_TOPIC("a1dc7", DEBUG, Logger::MAINTENANCE) << "SynchronizeOneShard: " << error.str();
        _result.reset(TRI_ERROR_FAILED, error.str());
        return false;
      }
  
      auto ci = clusterInfo->getCollectionNT(database, planId);
      if (ci == nullptr) {
        std::stringstream msg;
        msg << "exception in getCollection, ";
        AppendShardInformationToMessage(database, shard, planId, startTime, msg);
        LOG_TOPIC("89972", DEBUG, Logger::MAINTENANCE) << "SynchronizeOneShard: " << msg.str();
        _result.reset(TRI_ERROR_FAILED, msg.str());
        return false;
      }
  
      std::string const cid = std::to_string(ci->id());
      std::shared_ptr<CollectionInfoCurrent> cic =
          ClusterInfo::instance()->getCollectionCurrent(database, cid);
      std::vector<std::string> current = cic->servers(shard);
  
      if (current.empty()) {
        LOG_TOPIC("b0ccf", DEBUG, Logger::MAINTENANCE)
            << "synchronizeOneShard: cancelled, no servers in 'Current'";
      } else if (current.front() == leader) {
        if (std::find(current.begin(), current.end(), ourselves) == current.end()) {
          break;  // start synchronization work
        }
        // We are already there, this is rather strange, but never mind:
        std::stringstream error;
        error << "already done, ";
        AppendShardInformationToMessage(database, shard, planId, startTime, error);
        LOG_TOPIC("4abcb", DEBUG, Logger::MAINTENANCE) << "SynchronizeOneShard: " << error.str();
        _result.reset(TRI_ERROR_FAILED, error.str());
        return false;
      }
  
      LOG_TOPIC("28600", DEBUG, Logger::MAINTENANCE)
          << "synchronizeOneShard: waiting for leader, " << database << "/"
          << shard << ", " << database << "/" << planId;
  
      std::this_thread::sleep_for(duration<double>(0.2));
    }
  
    // Once we get here, we know that the leader is ready for sync, so we give it
    // a try:
  
    try {
      DatabaseGuard guard(database);
      auto vocbase = &guard.database();
  
      auto collection = vocbase->lookupCollection(shard);
      if (collection == nullptr) {
        std::stringstream error;
        error << "failed to lookup local shard " << shard;
        LOG_TOPIC("06489", ERR, Logger::MAINTENANCE) << "SynchronizeOneShard: " << error.str();
        _result.reset(TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND, error.str());
        return false;
      }
  
      auto ep = clusterInfo->getServerEndpoint(leader);
      uint64_t docCount;
      if (!collectionCount(collection, docCount).ok()) {
        std::stringstream error;
        error << "failed to get a count on leader " << shard;
        LOG_TOPIC("da225", ERR, Logger::MAINTENANCE) << "SynchronizeShard " << error.str();
        _result.reset(TRI_ERROR_INTERNAL, error.str());
        return false;
      }
  
      { // Initialize _clientInfoString
        CollectionNameResolver resolver(collection->vocbase());
        _clientInfoString =
            std::string{"follower "} + ServerState::instance()->getPersistedId() +
            " of shard " + collection->name() + " of collection " + database +
            "/" + resolver.getCollectionName(collection->id());
      }
  
      if (docCount == 0) {
        // We have a short cut:
        LOG_TOPIC("0932a", DEBUG, Logger::MAINTENANCE)
            << "synchronizeOneShard: trying short cut to synchronize local shard "
               "'"
            << database << "/" << shard << "' for central '" << database << "/"
            << planId << "'";
        try {
          auto asResult =
              addShardFollower(ep, database, shard, 0, clientId, SyncerId{}, _clientInfoString, 60.0);
  
          if (asResult.ok()) {
            if (Logger::isEnabled(LogLevel::DEBUG, Logger::MAINTENANCE)) {
              std::stringstream msg;
              msg << "synchronizeOneShard: shortcut worked, done, ";
              AppendShardInformationToMessage(database, shard, planId, startTime, msg);
              LOG_TOPIC("f4a5b", DEBUG, Logger::MAINTENANCE) << msg.str();
            }
            collection->followers()->setTheLeader(leader);
            notify();
            return false;
          }
        } catch (...) {
        }
      }
  
      LOG_TOPIC("53337", DEBUG, Logger::MAINTENANCE)
          << "synchronizeOneShard: trying to synchronize local shard '" << database
          << "/" << shard << "' for central '" << database << "/" << planId << "'";
  
      try {
        // From here on we perform a number of steps, each of which can
        // fail. If it fails with an exception, it is caught, but this
        // should usually not happen. If it fails without an exception,
        // we log and use return.
  
        Result res;  // used multiple times for intermediate results
  
        // First once without a read transaction:
  
        if (isStopping()) {
          std::string errorMessage(
              "synchronizeOneShard: synchronization failed for shard ");
          errorMessage += shard + ": shutdown in progress, giving up";
          LOG_TOPIC("a0f9a", INFO, Logger::MAINTENANCE) << errorMessage;
          _result.reset(TRI_ERROR_SHUTTING_DOWN, errorMessage);
          return false;
        }
  
        // This is necessary to accept replications from the leader which can
        // happen as soon as we are in sync.
        collection->followers()->setTheLeader(leader);
  
        startTime = system_clock::now();
  
        VPackBuilder config;
        {
          VPackObjectBuilder o(&config);
          config.add(ENDPOINT, VPackValue(ep));
          config.add(INCREMENTAL,
                     VPackValue(docCount > 0));  // use dump if possible
          config.add(KEEP_BARRIER, VPackValue(true));
          config.add(LEADER_ID, VPackValue(leader));
          config.add(SKIP_CREATE_DROP, VPackValue(true));
          config.add(RESTRICT_TYPE, VPackValue(INCLUDE));
          config.add(VPackValue(RESTRICT_COLLECTIONS));
          {
            VPackArrayBuilder a(&config);
            config.add(VPackValue(shard));
          }
          config.add(INCLUDE_SYSTEM, VPackValue(true));
          config.add("verbose", VPackValue(false));
        }
  
        auto details = std::make_shared<VPackBuilder>();
  
        ResultT<SyncerId> syncRes =
            replicationSynchronize(collection, config.slice(), _clientInfoString,
                                   APPLIER_DATABASE, details);
  
        auto sy = details->slice();
        auto const endTime = system_clock::now();
  
        // Long shard sync initialization
        if (endTime - startTime > seconds(5)) {
          LOG_TOPIC("ca7e3", INFO, Logger::MAINTENANCE)
              << "synchronizeOneShard: long call to syncCollection for shard"
              << shard << " " << syncRes.errorMessage()
              << " start time: " << timepointToString(startTime)
              << ", end time: " << timepointToString(system_clock::now());
        }
  
        // If this did not work, then we cannot go on:
        if (!syncRes.ok()) {
          std::stringstream error;
          error << "could not initially synchronize shard " << shard << ": "
                << syncRes.errorMessage();
          LOG_TOPIC("c1b31", DEBUG, Logger::MAINTENANCE) << "SynchronizeOneShard: " << error.str();
          _result.reset(TRI_ERROR_INTERNAL, error.str());
          return false;
        }
  
        SyncerId syncerId = syncRes.get();
        // From here on, we have to call `cancelBarrier` in case of errors
        // as well as in the success case!
        auto barrierId = sy.get(BARRIER_ID).getNumber<int64_t>();
        TRI_DEFER(cancelBarrier(ep, database, barrierId, clientId));
  
        VPackSlice collections = sy.get(COLLECTIONS);
  
        if (collections.length() == 0 || collections[0].get("name").copyString() != shard) {
          std::stringstream error;
          error
              << "shard " << shard
              << " seems to be gone from leader, this "
                 "can happen if a collection was dropped during synchronization!";
          LOG_TOPIC("664ae", WARN, Logger::MAINTENANCE) << "SynchronizeOneShard: " << error.str();
          _result.reset(TRI_ERROR_INTERNAL, error.str());
          return false;
        }
  
        auto lastTick =
            arangodb::basics::VelocyPackHelper::readNumericValue<TRI_voc_tick_t>(sy, LAST_LOG_TICK,
                                                                                 0);
        VPackBuilder builder;
  
        ResultT<TRI_voc_tick_t> tickResult =
            catchupWithReadLock(ep, database, *collection, clientId, shard,
                                leader, lastTick, builder);
        if (!tickResult.ok()) {
          LOG_TOPIC("0a4d4", INFO, Logger::MAINTENANCE) << syncRes.errorMessage();
          _result.reset(tickResult.result());
          return false;
        }
        lastTick = tickResult.get();
  
        // Now start a exclusive transaction to stop writes:
        res = catchupWithExclusiveLock(ep, database, *collection, clientId, shard,
                                       leader, syncerId, lastTick, builder);
        if (!res.ok()) {
          LOG_TOPIC("be85f", INFO, Logger::MAINTENANCE) << res.errorMessage();
          _result.reset(res);
          return false;
        }
  
      } catch (std::exception const& e) {
        std::stringstream error;
        error << "synchronization of";
        AppendShardInformationToMessage(database, shard, planId, startTime, error);
        error << " failed: " << e.what();
        LOG_TOPIC("1e576", ERR, Logger::MAINTENANCE) << error.str();
        _result.reset(TRI_ERROR_INTERNAL, e.what());
        return false;
      }
      // Validate that HARDLOCK only works!
    } catch (std::exception const& e) {
      // This catches the case that we could not even find the collection
      // locally, because the DatabaseGuard constructor threw.
      LOG_TOPIC("9f2c0", WARN, Logger::MAINTENANCE)
          << "action " << _description << " failed with exception " << e.what();
      _result.reset(TRI_ERROR_INTERNAL, e.what());
      return false;
    }
  
    // Tell others that we are done:
    if (Logger::isEnabled(LogLevel::INFO, Logger::MAINTENANCE)) {
      // This wrap is just to not write the stream if not needed.
      std::stringstream msg;
      AppendShardInformationToMessage(database, shard, planId, startTime, msg);
      LOG_TOPIC("e6780", INFO, Logger::MAINTENANCE) << "synchronizeOneShard: done, " << msg.str();
    }
    notify();
    return false;
  }
  
  ResultT<TRI_voc_tick_t> SynchronizeShard::catchupWithReadLock(
      std::string const& ep, std::string const& database, LogicalCollection const& collection,
      std::string const& clientId, std::string const& shard,
      std::string const& leader, TRI_voc_tick_t lastLogTick, VPackBuilder& builder) {
    bool didTimeout = true;
    int tries = 0;
    double timeout = 300.0;
    TRI_voc_tick_t tickReached = 0;
    while (didTimeout && tries++ < 18) {  // This will try to sync for at most 1 hour. (200 * 18 == 3600)
  
      if (isStopping()) {
        std::string errorMessage =
            "synchronizeOneShard: startReadLockOnLeader (soft): shutting down";
        return ResultT<TRI_voc_tick_t>::error(TRI_ERROR_SHUTTING_DOWN, errorMessage);
      }
  
      didTimeout = false;
      // Now ask for a "soft stop" on the leader, in case of mmfiles, this
      // will be a hard stop, but for rocksdb, this is a no-op:
      uint64_t lockJobId = 0;
      LOG_TOPIC("b4f2b", DEBUG, Logger::MAINTENANCE)
          << "synchronizeOneShard: startReadLockOnLeader (soft): " << ep << ":"
          << database << ":" << collection.name();
      Result res = startReadLockOnLeader(ep, database, collection.name(),
                                         clientId, lockJobId, true, timeout);
      if (!res.ok()) {
        std::string errorMessage =
            "synchronizeOneShard: error in startReadLockOnLeader (soft):" + res.errorMessage();
        return ResultT<TRI_voc_tick_t>::error(TRI_ERROR_INTERNAL, errorMessage);
      }
  
      auto readLockGuard = arangodb::scopeGuard([&]() {
        // Always cancel the read lock.
        // Reported seperately
        auto res = cancelReadLockOnLeader(ep, database, lockJobId, clientId, 60.0);
        if (!res.ok()) {
          LOG_TOPIC("b15ee", INFO, Logger::MAINTENANCE)
              << "Could not cancel soft read lock on leader: " << res.errorMessage();
        }
      });
  
      LOG_TOPIC("5eb37", DEBUG, Logger::MAINTENANCE) << "lockJobId: " << lockJobId;
  
      // From now on, we need to cancel the read lock on the leader regardless
      // if things go wrong or right!
  
      // Do a first try of a catch up with the WAL. In case of RocksDB,
      // this has not yet stopped the writes, so we have to be content
      // with nearly reaching the end of the WAL, which is a "soft" catchup.
      builder.clear();
      {
        VPackObjectBuilder o(&builder);
        builder.add(ENDPOINT, VPackValue(ep));
        builder.add(DATABASE, VPackValue(database));
        builder.add(COLLECTION, VPackValue(shard));
        builder.add(LEADER_ID, VPackValue(leader));
        builder.add("from", VPackValue(lastLogTick));
        builder.add("requestTimeout", VPackValue(600.0));
        builder.add("connectTimeout", VPackValue(60.0));
      }
  
      // We only allow to hold this lock for 60% of the timeout time, so to avoid
      // any issues with Locks timeouting on the Leader and the Client not
      // recognizing it.
      res = replicationSynchronizeCatchup(builder.slice(), timeout * 0.6, tickReached, didTimeout);
  
      if (!res.ok()) {
        std::string errorMessage(
            "synchronizeOneshard: error in syncCollectionCatchup: ");
        errorMessage += res.errorMessage();
        return ResultT<TRI_voc_tick_t>::error(TRI_ERROR_INTERNAL, errorMessage);
      }
  
      // Stop the read lock again:
      res = cancelReadLockOnLeader(ep, database, lockJobId, clientId, 60.0);
      // We removed the readlock
      readLockGuard.cancel();
      if (!res.ok()) {
        std::string errorMessage =
            "synchronizeOneShard: error when cancelling soft read lock: " + res.errorMessage();
        LOG_TOPIC("c37d1", INFO, Logger::MAINTENANCE) << errorMessage;
        _result.reset(TRI_ERROR_INTERNAL, errorMessage);
        return ResultT<TRI_voc_tick_t>::error(TRI_ERROR_INTERNAL, errorMessage);
      }
      lastLogTick = tickReached;
      if (didTimeout) {
        LOG_TOPIC("e516e", INFO, Logger::MAINTENANCE)
            << "Renewing softLock for " << shard << " on leader: " << leader;
      }
    }
    if (didTimeout) {
      LOG_TOPIC("f1a61", WARN, Logger::MAINTENANCE)
          << "Could not catchup under softLock for " << shard << " on leader: " << leader
          << " now activating hardLock. This is expected under high load.";
    }
    return ResultT<TRI_voc_tick_t>::success(tickReached);
  }
  
  Result SynchronizeShard::catchupWithExclusiveLock(
      std::string const& ep, std::string const& database, LogicalCollection const& collection,
      std::string const& clientId, std::string const& shard, std::string const& leader,
      SyncerId const syncerId, TRI_voc_tick_t lastLogTick, VPackBuilder& builder) {
    uint64_t lockJobId = 0;
    LOG_TOPIC("da129", DEBUG, Logger::MAINTENANCE)
        << "synchronizeOneShard: startReadLockOnLeader: " << ep << ":" << database
        << ":" << collection.name();
    Result res = startReadLockOnLeader(ep, database, collection.name(), clientId,
                                       lockJobId, false);
    if (!res.ok()) {
      std::string errorMessage =
          "synchronizeOneShard: error in startReadLockOnLeader (hard):" + res.errorMessage();
      return {TRI_ERROR_INTERNAL, errorMessage};
    }
    auto readLockGuard = arangodb::scopeGuard([&]() {
      // Always cancel the read lock.
      // Reported seperately
      auto res = cancelReadLockOnLeader(ep, database, lockJobId, clientId, 60.0);
      if (!res.ok()) {
        LOG_TOPIC("067a8", INFO, Logger::MAINTENANCE)
            << "Could not cancel hard read lock on leader: " << res.errorMessage();
      }
    });
  
    LOG_TOPIC("d76cb", DEBUG, Logger::MAINTENANCE) << "lockJobId: " << lockJobId;
  
    builder.clear();
    {
      VPackObjectBuilder o(&builder);
      builder.add(ENDPOINT, VPackValue(ep));
      builder.add(DATABASE, VPackValue(database));
      builder.add(COLLECTION, VPackValue(shard));
      builder.add(LEADER_ID, VPackValue(leader));
      builder.add("from", VPackValue(lastLogTick));
      builder.add("requestTimeout", VPackValue(600.0));
      builder.add("connectTimeout", VPackValue(60.0));
    }
  
    res = replicationSynchronizeFinalize(builder.slice());
  
    if (!res.ok()) {
      std::string errorMessage(
          "synchronizeOneshard: error in syncCollectionFinalize: ");
      errorMessage += res.errorMessage();
      return {TRI_ERROR_INTERNAL, errorMessage};
    }
  
    res = addShardFollower(ep, database, shard, lockJobId, clientId, syncerId, _clientInfoString, 60.0);
  
    if (!res.ok()) {
      std::string errorMessage(
          "synchronizeOneshard: error in addShardFollower: ");
      errorMessage += res.errorMessage();
      return {TRI_ERROR_INTERNAL, errorMessage};
    }
  
    // Report success:
    LOG_TOPIC("3423d", DEBUG, Logger::MAINTENANCE)
        << "synchronizeOneShard: synchronization worked for shard " << shard;
    _result.reset(TRI_ERROR_NO_ERROR);
    return {TRI_ERROR_NO_ERROR};
  }
  
  void SynchronizeShard::setState(ActionState state) {
    if ((COMPLETE == state || FAILED == state) && _state != state) {
      TRI_ASSERT(_description.has("shard"));
      _feature.incShardVersion(_description.get("shard"));
    }
  
    ActionBase::setState(state);
  }
  