/// Load persisted configuration
bool State::loadOrPersistConfiguration() {
  auto bindVars = std::make_shared<VPackBuilder>();
  bindVars->openObject();
  bindVars->close();

  std::string const aql(
      std::string("FOR c in configuration FILTER c._key==\"0\" RETURN c.cfg"));

  arangodb::aql::Query query(false, _vocbase, aql.c_str(), aql.size(), bindVars,
                             nullptr, arangodb::aql::PART_MAIN);

  auto queryResult = query.execute(QueryRegistryFeature::QUERY_REGISTRY);

  if (queryResult.code != TRI_ERROR_NO_ERROR) {
    THROW_ARANGO_EXCEPTION_MESSAGE(queryResult.code, queryResult.details);
  }

  VPackSlice result = queryResult.result->slice();

  if (result.isArray() &&
      result.length()) {  // We already have a persisted conf

    try {
      LOG_TOPIC(DEBUG, Logger::AGENCY)
        << "Merging configuration " << result[0].resolveExternals().toJson();
      _agent->mergeConfiguration(result[0].resolveExternals());
      
    } catch (std::exception const& e) {
      LOG_TOPIC(ERR, Logger::AGENCY)
          << "Failed to merge persisted configuration into runtime "
             "configuration: "
          << e.what();
      FATAL_ERROR_EXIT();
    }

  } else {  // Fresh start

    LOG_TOPIC(DEBUG, Logger::AGENCY) << "New agency!";

    TRI_ASSERT(_agent != nullptr);
    _agent->id(to_string(boost::uuids::random_generator()()));

    auto transactionContext =
        std::make_shared<StandaloneTransactionContext>(_vocbase);
    SingleCollectionTransaction trx(transactionContext, "configuration",
                                    TRI_TRANSACTION_WRITE);

    int res = trx.begin();
    OperationResult result;

    if (res != TRI_ERROR_NO_ERROR) {
      THROW_ARANGO_EXCEPTION(res);
    }

    Builder doc;
    doc.openObject();
    doc.add("_key", VPackValue("0"));
    doc.add("cfg", _agent->config().toBuilder()->slice());
    doc.close();
    try {
      result = trx.insert("configuration", doc.slice(), _options);
    } catch (std::exception const& e) {
      LOG_TOPIC(ERR, Logger::AGENCY) << "Failed to persist configuration entry:"
                                     << e.what();
      FATAL_ERROR_EXIT();
    }

    res = trx.finish(result.code);

    return (res == TRI_ERROR_NO_ERROR);
  }

  return true;
}

ShortestPathOptions::ShortestPathOptions(VPackSlice const& slice) {
  VPackSlice obj = slice.get("shortestpathFlags");
  
  weightAttribute = "";
  if (obj.hasKey("weightAttribute")) {
    VPackSlice v = obj.get("weightAttribute");
    if (v.isString()) {
      weightAttribute = v.copyString();
    }
  }
  
  defaultWeight = 1;
  if (obj.hasKey("defaultWeight")) {
    VPackSlice v = obj.get("defaultWeight");
    if (v.isNumber()) {
      defaultWeight = v.getNumericValue<double>();
    }
  }
}

v8::Handle<v8::Value> TRI_VPackToV8(v8::Isolate* isolate,
                                    VPackSlice const& slice,
                                    VPackOptions const* options,
                                    VPackSlice const* base) {
  switch (slice.type()) {
    case VPackValueType::Null: {
      return v8::Null(isolate);
    }
    case VPackValueType::Bool: {
      return v8::Boolean::New(isolate, slice.getBool());
    }
    case VPackValueType::Double: {
      // convert NaN, +inf & -inf to null
      double value = slice.getDouble();
      if (std::isnan(value) || !std::isfinite(value) || value == HUGE_VAL || value == -HUGE_VAL) {
        return v8::Null(isolate);
      }
      return v8::Number::New(isolate, slice.getDouble());
    }
    case VPackValueType::Int: {
      int64_t value = slice.getInt();
      if (value >= -2147483648LL && value <= 2147483647LL) {
        // value is within bounds of an int32_t
        return v8::Integer::New(isolate, static_cast<int32_t>(value));
      }
      if (value >= 0 && value <= 4294967295LL) {
        // value is within bounds of a uint32_t
        return v8::Integer::NewFromUnsigned(isolate, static_cast<uint32_t>(value));
      }
      // must use double to avoid truncation
      return v8::Number::New(isolate, static_cast<double>(slice.getInt()));
    }
    case VPackValueType::UInt: {
      uint64_t value = slice.getUInt();
      if (value <= 4294967295ULL) {
        // value is within bounds of a uint32_t
        return v8::Integer::NewFromUnsigned(isolate, static_cast<uint32_t>(value));
      }
      // must use double to avoid truncation
      return v8::Number::New(isolate, static_cast<double>(slice.getUInt()));
    }
    case VPackValueType::SmallInt: {
      return v8::Integer::New(isolate, slice.getNumericValue<int32_t>());
    }
    case VPackValueType::String: {
      return ObjectVPackString(isolate, slice);
    }
    case VPackValueType::Array: {
      return ObjectVPackArray(isolate, slice, options, base);
    }
    case VPackValueType::Object: {
      return ObjectVPackObject(isolate, slice, options, base);
    }
    case VPackValueType::External: {
      // resolve external
      return TRI_VPackToV8(isolate, VPackSlice(slice.getExternal()), options, base);
    }
    case VPackValueType::Custom: {
      if (options == nullptr || options->customTypeHandler == nullptr || base == nullptr) {
        THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL,
                                       "Could not extract custom attribute.");
      }
      std::string id =
          options->customTypeHandler->toString(slice, options, *base);
      return TRI_V8_STD_STRING(id);
    }
    case VPackValueType::None:
    default: {
      return v8::Undefined(isolate);
    }
  }
}

/// @brief Call to election
void Constituent::callElection() {
  std::vector<bool> votes(size(), false);

  votes.at(_id) = true;  // vote for myself
  _cast = true;
  if (_role == CANDIDATE) {
    this->term(_term + 1);  // raise my term
  }

  std::string body;
  std::vector<OperationID> operationIDs(config().endpoints.size());
  std::stringstream path;

  path << "/_api/agency_priv/requestVote?term=" << _term
       << "&candidateId=" << _id << "&prevLogIndex=" << _agent->lastLog().index
       << "&prevLogTerm=" << _agent->lastLog().term;

  // Ask everyone for their vote
  for (arangodb::consensus::id_t i = 0; i < config().endpoints.size(); ++i) {
    if (i != _id && endpoint(i) != "") {
      auto headerFields =
          std::make_unique<std::unordered_map<std::string, std::string>>();
      operationIDs[i] = arangodb::ClusterComm::instance()->asyncRequest(
          "1", 1, config().endpoints[i], GeneralRequest::RequestType::GET,
          path.str(), std::make_shared<std::string>(body), headerFields,
          nullptr, config().minPing, true);
    }
  }

  // Wait randomized timeout
  std::this_thread::sleep_for(
      sleepFor(.5 * config().minPing, .8 * config().minPing));

  // Collect votes
  // FIXME: This code can be improved: One can wait for an arbitrary
  // result by creating a coordinatorID and waiting for a pattern.
  for (arangodb::consensus::id_t i = 0; i < config().endpoints.size(); ++i) {
    if (i != _id && endpoint(i) != "") {
      ClusterCommResult res =
          arangodb::ClusterComm::instance()->enquire(operationIDs[i]);

      if (res.status == CL_COMM_SENT) {  // Request successfully sent
        res = arangodb::ClusterComm::instance()->wait(
            "1", 1, operationIDs[i], "1");
        std::shared_ptr<Builder> body = res.result->getBodyVelocyPack();
        if (body->isEmpty()) {  // body empty
          continue;
        } else {
          if (body->slice().isObject()) {  // body
            VPackSlice slc = body->slice();
            if (slc.hasKey("term") && slc.hasKey("voteGranted")) {  // OK
              term_t t = slc.get("term").getUInt();
              if (t > _term) {  // follow?
                follow(t);
                break;
              }
              votes[i] = slc.get("voteGranted").getBool();  // Get vote
            }
          }
        }
      } else {  // Request failed
        votes[i] = false;
      }
    }
  }

  // Count votes
  size_t yea = 0;
  for (size_t i = 0; i < size(); ++i) {
    if (votes[i]) {
      yea++;
    }
  }

  // Evaluate election results
  if (yea > size() / 2) {
    lead(votes);
  } else {
    follow(_term);
  }
  
}

int Syncer::handleStateResponse(VPackSlice const& slice, std::string& errorMsg) {
  std::string const endpointString =
      " from endpoint '" + _masterInfo._endpoint + "'";

  // process "state" section
  VPackSlice const state = slice.get("state");

  if (!state.isObject()) {
    errorMsg = "state section is missing in response" + endpointString;

    return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
  }

  // state."lastLogTick"
  VPackSlice const tick = state.get("lastLogTick");

  if (!tick.isString()) {
    errorMsg = "lastLogTick is missing in response" + endpointString;

    return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
  }

  TRI_voc_tick_t const lastLogTick = VelocyPackHelper::stringUInt64(tick);

  // state."running"
  bool running = VelocyPackHelper::getBooleanValue(state, "running", false);

  // process "server" section
  VPackSlice const server = slice.get("server");

  if (!server.isObject()) {
    errorMsg = "server section is missing in response" + endpointString;

    return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
  }

  // server."version"
  VPackSlice const version = server.get("version");

  if (!version.isString()) {
    errorMsg = "server version is missing in response" + endpointString;

    return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
  }

  // server."serverId"
  VPackSlice const serverId = server.get("serverId");

  if (!serverId.isString()) {
    errorMsg = "server id is missing in response" + endpointString;

    return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
  }

  // validate all values we got
  std::string const masterIdString(serverId.copyString());
  TRI_server_id_t const masterId = StringUtils::uint64(masterIdString);

  if (masterId == 0) {
    // invalid master id
    errorMsg = "invalid server id in response" + endpointString;

    return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
  }

  if (masterIdString == _localServerIdString) {
    // master and replica are the same instance. this is not supported.
    errorMsg = "got same server id (" + _localServerIdString + ")" +
               endpointString + " as the local applier server's id";

    return TRI_ERROR_REPLICATION_LOOP;
  }

  int major = 0;
  int minor = 0;

  std::string const versionString(version.copyString());

  if (sscanf(versionString.c_str(), "%d.%d", &major, &minor) != 2) {
    errorMsg = "invalid master version info" + endpointString + ": '" +
               versionString + "'";

    return TRI_ERROR_REPLICATION_MASTER_INCOMPATIBLE;
  }

  if (major != 3) {
    // we can connect to 3.x only
    errorMsg = "got incompatible master version" + endpointString + ": '" +
               versionString + "'";

    return TRI_ERROR_REPLICATION_MASTER_INCOMPATIBLE;
  }

  _masterInfo._majorVersion = major;
  _masterInfo._minorVersion = minor;
  _masterInfo._serverId = masterId;
  _masterInfo._lastLogTick = lastLogTick;
  _masterInfo._active = running;

  LOG_TOPIC(INFO, Logger::REPLICATION)
//.........这里部分代码省略.........

/// @brief Call to election
void Constituent::callElection() {
  std::map<std::string, bool> votes;
  std::vector<std::string> active =
      _agent->config().active();  // Get copy of active

  votes[_id] = true;  // vote for myself

  term_t savedTerm;
  {
    MUTEX_LOCKER(locker, _castLock);
    this->termNoLock(_term + 1);  // raise my term
    _cast = true;
    _votedFor = _id;
    savedTerm = _term;
    LOG_TOPIC(DEBUG, Logger::AGENCY) << "Set _leaderID to NO_LEADER"
      << " in term " << _term;
    _leaderID = NO_LEADER;
  }

  std::string body;
  std::map<std::string, OperationID> operationIDs;
  std::stringstream path;

  path << "/_api/agency_priv/requestVote?term=" << savedTerm
       << "&candidateId=" << _id << "&prevLogIndex=" << _agent->lastLog().index
       << "&prevLogTerm=" << _agent->lastLog().term;

  double minPing = _agent->config().minPing();

  double respTimeout = 0.9 * minPing;
  double initTimeout = 0.5 * minPing;

  // Ask everyone for their vote
  for (auto const& i : active) {
    if (i != _id) {
      auto headerFields =
          std::make_unique<std::unordered_map<std::string, std::string>>();
      operationIDs[i] = ClusterComm::instance()->asyncRequest(
        "1", 1, _agent->config().poolAt(i),
        rest::RequestType::GET, path.str(),
        std::make_shared<std::string>(body), headerFields,
        nullptr, respTimeout, true, initTimeout);
    }
  }

  // Wait randomized timeout
  std::this_thread::sleep_for(sleepFor(initTimeout, respTimeout));

  // Collect votes
  for (const auto& i : active) {
    if (i != _id) {
      ClusterCommResult res =
          arangodb::ClusterComm::instance()->enquire(operationIDs[i]);
      if (res.status == CL_COMM_SENT) {  // Request successfully sent
        res = ClusterComm::instance()->wait("1", 1, operationIDs[i], "1");
        std::shared_ptr<Builder> body = res.result->getBodyVelocyPack();
        if (body->isEmpty()) {  // body empty
          continue;
        } else {
          if (body->slice().isObject()) {  // body
            VPackSlice slc = body->slice();
            if (slc.hasKey("term") && slc.hasKey("voteGranted")) {  // OK
              term_t t = slc.get("term").getUInt();
              if (t > _term) {  // follow?
                follow(t);
                break;
              }
              votes[i] = slc.get("voteGranted").getBool();  // Get vote
            }
          }
        }
      } else {  // Request failed
        votes[i] = false;
      }
    }
  }

  // Count votes
  size_t yea = 0;
  for (auto const& i : votes) {
    if (i.second) {
      ++yea;
    }
  }

  {
    MUTEX_LOCKER(locker, _castLock);
    if (savedTerm != _term) {
      followNoLock(_term);
      return;
    }
  }

  // Evaluate election results
  if (yea > size() / 2) {
    lead(savedTerm, votes);
  } else {
    follow(_term);
  }
//.........这里部分代码省略.........

/// @brief create the block from VelocyPack, note that this can throw
AqlItemBlock::AqlItemBlock(VPackSlice const slice) {
  bool exhausted = VelocyPackHelper::getBooleanValue(slice, "exhausted", false);

  if (exhausted) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL,
                                   "exhausted must be false");
  }

  _nrItems = VelocyPackHelper::getNumericValue<size_t>(slice, "nrItems", 0);
  if (_nrItems == 0) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL, "nrItems must be > 0");
  }

  _nrRegs = VelocyPackHelper::getNumericValue<RegisterId>(slice, "nrRegs", 0);

  // Initialize the data vector:
  if (_nrRegs > 0) {
    _data.resize(_nrItems * _nrRegs);
  }

  // Now put in the data:
  VPackSlice data = slice.get("data");
  VPackSlice raw = slice.get("raw");

  std::vector<AqlValue> madeHere;
  madeHere.reserve(static_cast<size_t>(raw.length()));
  madeHere.emplace_back();  // an empty AqlValue
  madeHere.emplace_back();  // another empty AqlValue, indices start w. 2

  try {
    size_t posInRaw = 2;
    size_t posInData = 0;
    int64_t emptyRun = 0;

    for (RegisterId column = 0; column < _nrRegs; column++) {
      for (size_t i = 0; i < _nrItems; i++) {
        if (emptyRun > 0) {
          emptyRun--;
        } else {
          VPackSlice dataEntry = data.at(posInData++);
          if (!dataEntry.isNumber()) {
            THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL,
                                           "data must contain only numbers");
          }
          int64_t n = dataEntry.getNumericValue<int64_t>();
          if (n == 0) {
            // empty, do nothing here
          } else if (n == -1) {
            // empty run:
            VPackSlice runLength = data.at(posInData++);
            TRI_ASSERT(runLength.isNumber());
            emptyRun = runLength.getNumericValue<int64_t>();
            emptyRun--;
          } else if (n == -2) {
            // a range
            VPackSlice lowBound = data.at(posInData++);
            VPackSlice highBound = data.at(posInData++);
            
            int64_t low =
                VelocyPackHelper::getNumericValue<int64_t>(lowBound, 0);
            int64_t high =
                VelocyPackHelper::getNumericValue<int64_t>(highBound, 0);
            AqlValue a(low, high);
            try {
              setValue(i, column, a);
            } catch (...) {
              a.destroy();
              throw;
            }
          } else if (n == 1) {
            // a VelocyPack value
            AqlValue a(raw.at(posInRaw++));
            try {
              setValue(i, column, a);  // if this throws, a is destroyed again
            } catch (...) {
              a.destroy();
              throw;
            }
            madeHere.emplace_back(a);
          } else if (n >= 2) {
            setValue(i, column, madeHere[static_cast<size_t>(n)]);
            // If this throws, all is OK, because it was already put into
            // the block elsewhere.
          } else {
            THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL,
                                           "found undefined data value");
          }
        }
      }
    }
  } catch (...) {
    destroy();
  }
}

arangodb::traverser::TraverserOptions::LookupInfo::LookupInfo(
    arangodb::aql::Query* query, VPackSlice const& info, VPackSlice const& shards) {
  TRI_ASSERT(shards.isArray());
  idxHandles.reserve(shards.length());

  conditionNeedUpdate = arangodb::basics::VelocyPackHelper::getBooleanValue(
      info, "condNeedUpdate", false);
  conditionMemberToUpdate =
      arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(
          info, "condMemberToUpdate", 0);
  
  VPackSlice read = info.get("handle");
  if (!read.isObject()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(
        TRI_ERROR_BAD_PARAMETER,
        "Each lookup requires handle to be an object");
  }

  read = read.get("id");
  if (!read.isString()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(
        TRI_ERROR_BAD_PARAMETER,
        "Each handle requires id to be a string");
  }
  std::string idxId = read.copyString();
  auto trx = query->trx();

  for (auto const& it : VPackArrayIterator(shards)) {
    if (!it.isString()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(
          TRI_ERROR_BAD_PARAMETER,
          "Shards have to be a list of strings");
    }
    idxHandles.emplace_back(trx->getIndexByIdentifier(it.copyString(), idxId));
  }

  read = info.get("expression");
  if (!read.isObject()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(
        TRI_ERROR_BAD_PARAMETER,
        "Each lookup requires expression to be an object");
  }


  expression = new aql::Expression(query->ast(), read);

  read = info.get("condition");
  if (!read.isObject()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(
        TRI_ERROR_BAD_PARAMETER,
        "Each lookup requires condition to be an object");
  }
  indexCondition = new aql::AstNode(query->ast(), read); 
}

ShortestPathNode::ShortestPathNode(ExecutionPlan* plan,
                                   arangodb::velocypack::Slice const& base)
    : ExecutionNode(plan, base),
      _vocbase(plan->getAst()->query()->vocbase()),
      _vertexOutVariable(nullptr),
      _edgeOutVariable(nullptr),
      _inStartVariable(nullptr),
      _inTargetVariable(nullptr),
      _graphObj(nullptr) {
  // Directions
  VPackSlice dirList = base.get("directions");
  for (auto const& it : VPackArrayIterator(dirList)) {
    uint64_t dir = arangodb::basics::VelocyPackHelper::stringUInt64(it);
    TRI_edge_direction_e d;
    switch (dir) {
      case 0:
        d = TRI_EDGE_ANY;
        break;
      case 1:
        d = TRI_EDGE_IN;
        break;
      case 2:
        d = TRI_EDGE_OUT;
        break;
      default:
        THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                       "Invalid direction value");
        break;
    }
    _directions.emplace_back(d);
  }

  // Start Vertex
  if (base.hasKey("startInVariable")) {
    _inStartVariable = varFromVPack(plan->getAst(), base, "startInVariable");
  } else {
    VPackSlice v = base.get("startVertexId");
    if (!v.isString()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_BAD_JSON_PLAN,
                                     "start vertex must be a string");
    }
    _startVertexId = v.copyString();

    if (_startVertexId.empty()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_BAD_JSON_PLAN,
                                     "start vertex mustn't be empty");
    }
  }

  // Target Vertex
  if (base.hasKey("targetInVariable")) {
    _inTargetVariable = varFromVPack(plan->getAst(), base, "targetInVariable");
  } else {
    VPackSlice v = base.get("targetVertexId");
    if (!v.isString()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_BAD_JSON_PLAN,
                                     "target vertex must be a string");
    }
    _targetVertexId = v.copyString();
    if (_targetVertexId.empty()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_BAD_JSON_PLAN,
                                     "target vertex mustn't be empty");
    }
  }

  std::string graphName;
  if (base.hasKey("graph") && (base.get("graph").isString())) {
    graphName = base.get("graph").copyString();
    if (base.hasKey("graphDefinition")) {
      _graphObj = plan->getAst()->query()->lookupGraphByName(graphName);

      if (_graphObj == nullptr) {
        THROW_ARANGO_EXCEPTION(TRI_ERROR_GRAPH_NOT_FOUND);
      }

      auto eColls = _graphObj->edgeCollections();
      for (auto const& n : eColls) {
        _edgeColls.push_back(n);
      }
    } else {
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_BAD_JSON_PLAN,
                                     "missing graphDefinition.");
    }
  } else {
    _graphInfo.add(base.get("graph"));
    if (!_graphInfo.slice().isArray()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_BAD_JSON_PLAN,
                                     "graph has to be an array.");
    }
    // List of edge collection names
    for (auto const& it : VPackArrayIterator(_graphInfo.slice())) {
      if (!it.isString()) {
        THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_BAD_JSON_PLAN,
                                       "graph has to be an array of strings.");
      }
      _edgeColls.emplace_back(it.copyString());
    }
    if (_edgeColls.empty()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(
          TRI_ERROR_QUERY_BAD_JSON_PLAN,
//.........这里部分代码省略.........

static void JS_JsonCursor(v8::FunctionCallbackInfo<v8::Value> const& args) {
  TRI_V8_TRY_CATCH_BEGIN(isolate);
  v8::HandleScope scope(isolate);

  TRI_vocbase_t* vocbase = GetContextVocBase(isolate);

  if (vocbase == nullptr) {
    TRI_V8_THROW_EXCEPTION(TRI_ERROR_ARANGO_DATABASE_NOT_FOUND);
  }

  if (args.Length() != 1) {
    TRI_V8_THROW_EXCEPTION_USAGE("JSON_CURSOR(<id>)");
  }

  std::string const id = TRI_ObjectToString(args[0]);
  auto cursorId = static_cast<arangodb::CursorId>(
      arangodb::basics::StringUtils::uint64(id));

  // find the cursor
  auto cursors = vocbase->cursorRepository();
  TRI_ASSERT(cursors != nullptr);

  bool busy;
  auto cursor = cursors->find(cursorId, Cursor::CURSOR_VPACK, busy);

  if (cursor == nullptr) {
    if (busy) {
      TRI_V8_THROW_EXCEPTION(TRI_ERROR_CURSOR_BUSY);
    }

    TRI_V8_THROW_EXCEPTION(TRI_ERROR_CURSOR_NOT_FOUND);
  }

  try {
    auto result = v8::Object::New(isolate);

    // build documents
    auto docs = v8::Array::New(isolate);
    size_t const n = cursor->batchSize();

    for (size_t i = 0; i < n; ++i) {
      if (!cursor->hasNext()) {
        break;
      }

      auto row = cursor->next();

      if (row.isNone()) {
        TRI_V8_THROW_EXCEPTION(TRI_ERROR_OUT_OF_MEMORY);
      }

      docs->Set(static_cast<uint32_t>(i), TRI_VPackToV8(isolate, row));
    }

    result->ForceSet(TRI_V8_ASCII_STRING("result"), docs);

    bool hasCount = cursor->hasCount();
    size_t count = cursor->count();
    bool hasNext = cursor->hasNext();
    VPackSlice const extra = cursor->extra();

    result->ForceSet(TRI_V8_ASCII_STRING("hasMore"),
                     v8::Boolean::New(isolate, hasNext));

    if (hasNext) {
      result->ForceSet(TRI_V8_ASCII_STRING("id"),
                       V8TickId(isolate, cursor->id()));
    }

    if (hasCount) {
      result->ForceSet(TRI_V8_ASCII_STRING("count"),
                       v8::Number::New(isolate, static_cast<double>(count)));
    }
    if (!extra.isNone()) {
      result->ForceSet(TRI_V8_ASCII_STRING("extra"),
                       TRI_VPackToV8(isolate, extra));
    }

    cursors->release(cursor);

    TRI_V8_RETURN(result);
  } catch (...) {
    cursors->release(cursor);
    TRI_V8_THROW_EXCEPTION_MEMORY();
  }
  TRI_V8_TRY_CATCH_END
}

arangodb::traverser::TraverserOptions::TraverserOptions(
    arangodb::aql::Query* query, VPackSlice info, VPackSlice collections)
    : _trx(query->trx()),
      _baseVertexExpression(nullptr),
      _tmpVar(nullptr),
      _ctx(new aql::FixedVarExpressionContext()),
      _isCoordinator(arangodb::ServerState::instance()->isCoordinator()),
      minDepth(1),
      maxDepth(1),
      useBreadthFirst(false),
      uniqueVertices(UniquenessLevel::NONE),
      uniqueEdges(UniquenessLevel::PATH) {
      // NOTE collections is an array of arrays of strings
  VPackSlice read = info.get("minDepth");
  if (!read.isInteger()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                   "The options require a minDepth");
  }
  minDepth = read.getNumber<uint64_t>();

  read = info.get("maxDepth");
  if (!read.isInteger()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                   "The options require a maxDepth");
  }
  maxDepth = read.getNumber<uint64_t>();

  read = info.get("bfs");
  if (!read.isBoolean()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                   "The options require a bfs");
  }
  useBreadthFirst = read.getBool();

  read = info.get("tmpVar");
  if (!read.isObject()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                   "The options require a tmpVar");
  }
  _tmpVar = query->ast()->variables()->createVariable(read);

  read = info.get("uniqueVertices");
  if (!read.isInteger()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                   "The options require a uniqueVertices");
  }
  size_t i = read.getNumber<size_t>();
  switch (i) {
    case 0:
      uniqueVertices = UniquenessLevel::NONE;
      break;
    case 1:
      uniqueVertices = UniquenessLevel::PATH;
      break;
    case 2:
      uniqueVertices = UniquenessLevel::GLOBAL;
      break;
    default:
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                     "The options require a uniqueVertices");
  }

  read = info.get("uniqueEdges");
  if (!read.isInteger()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                   "The options require a uniqueEdges");
  }
  i = read.getNumber<size_t>();
  switch (i) {
    case 0:
      uniqueEdges = UniquenessLevel::NONE;
      break;
    case 1:
      uniqueEdges = UniquenessLevel::PATH;
      break;
    case 2:
      uniqueEdges = UniquenessLevel::GLOBAL;
      break;
    default:
      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                     "The options require a uniqueEdges");
  }

  read = info.get("baseLookupInfos");
  if (!read.isArray()) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_BAD_PARAMETER,
                                   "The options require a baseLookupInfos");
  }

  size_t length = read.length();
  TRI_ASSERT(read.length() == collections.length());
  _baseLookupInfos.reserve(length);
  for (size_t j = 0; j < length; ++j) {
    _baseLookupInfos.emplace_back(query, read.at(j), collections.at(j));
  }

  read = info.get("depthLookupInfo");
  if (!read.isNone()) {
    if (!read.isObject()) {
      THROW_ARANGO_EXCEPTION_MESSAGE(
//.........这里部分代码省略.........

void VelocyPackCursor::dump(arangodb::basics::StringBuffer& buffer) {
  buffer.appendText("\"result\":[");

  size_t const n = batchSize();

  // reserve 48 bytes per result document by default, but only
  // if the specified batch size does not get out of hand
  // otherwise specifying a very high batch size would make the allocation fail
  // in every case, even if there were much less documents in the collection
  size_t num = n;
  if (num == 0) {
    num = 1;
  } else if (num >= 10000) {
    num = 10000;
  }
  int res = buffer.reserve(num * 48);

  if (res != TRI_ERROR_NO_ERROR) {
    THROW_ARANGO_EXCEPTION(res);
  }

  arangodb::basics::VelocyPackDumper dumper(&buffer, _result.context->getVPackOptions());

  try {

    for (size_t i = 0; i < n; ++i) {
      if (!hasNext()) {
        break;
      }

      if (i > 0) {
        buffer.appendChar(',');
      }

      auto row = next();

      if (row.isNone()) {
        THROW_ARANGO_EXCEPTION(TRI_ERROR_OUT_OF_MEMORY);
      }

      dumper.dumpValue(row);
    }
  } catch (arangodb::basics::Exception const& ex) {
    THROW_ARANGO_EXCEPTION_MESSAGE(ex.code(), ex.what());
  } catch (std::exception const& ex) {
    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL, ex.what());
  } catch (...) {
    THROW_ARANGO_EXCEPTION(TRI_ERROR_INTERNAL);
  }

  buffer.appendText("],\"hasMore\":");
  buffer.appendText(hasNext() ? "true" : "false");

  if (hasNext()) {
    // only return cursor id if there are more documents
    buffer.appendText(",\"id\":\"");
    buffer.appendInteger(id());
    buffer.appendText("\"");
  }

  if (hasCount()) {
    buffer.appendText(",\"count\":");
    buffer.appendInteger(static_cast<uint64_t>(count()));
  }

  VPackSlice const extraSlice = extra();

  if (extraSlice.isObject()) {
    buffer.appendText(",\"extra\":");
    dumper.dumpValue(extraSlice);
  }

  buffer.appendText(",\"cached\":");
  buffer.appendText(_cached ? "true" : "false");

  if (!hasNext()) {
    // mark the cursor as deleted
    this->deleted();
  }
}

void RestWalHandler::properties() {
  auto l = arangodb::wal::LogfileManager::instance();

  if (_request->requestType() == rest::RequestType::PUT) {
    std::shared_ptr<VPackBuilder> parsedRequest;
    VPackSlice slice;
    try {
      slice = _request->payload();
    } catch (...) {
      generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                    "invalid body value. expecting object");
      return;
    }
    if (!slice.isObject()) {
      generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                    "invalid body value. expecting object");
    }

    if (slice.hasKey("allowOversizeEntries")) {
      bool value = slice.get("allowOversizeEntries").getBoolean();
      l->allowOversizeEntries(value);
    }
    
    if (slice.hasKey("logfileSize")) {
      uint32_t value = slice.get("logfileSize").getNumericValue<uint32_t>();
      l->filesize(value);
    }
    
    if (slice.hasKey("historicLogfiles")) {
      uint32_t value = slice.get("historicLogfiles").getNumericValue<uint32_t>();
      l->historicLogfiles(value);
    }
    
    if (slice.hasKey("reserveLogfiles")) {
      uint32_t value = slice.get("reserveLogfiles").getNumericValue<uint32_t>();
      l->reserveLogfiles(value);
    }
    
    if (slice.hasKey("throttleWait")) {
      uint64_t value = slice.get("throttleWait").getNumericValue<uint64_t>();
      l->maxThrottleWait(value);
    }
    
    if (slice.hasKey("throttleWhenPending")) {
      uint64_t value = slice.get("throttleWhenPending").getNumericValue<uint64_t>();
      l->throttleWhenPending(value);
    }
  }

  VPackBuilder builder;
  builder.openObject();
  builder.add("allowOversizeEntries", VPackValue(l->allowOversizeEntries()));
  builder.add("logfileSize", VPackValue(l->filesize()));
  builder.add("historicLogfiles", VPackValue(l->historicLogfiles()));
  builder.add("reserveLogfiles", VPackValue(l->reserveLogfiles()));
  builder.add("syncInterval", VPackValue(l->syncInterval()));
  builder.add("throttleWait", VPackValue(l->maxThrottleWait()));
  builder.add("throttleWhenPending", VPackValue(l->throttleWhenPending()));

  builder.close();
  generateResult(rest::ResponseCode::OK, builder.slice());
}

void RestWalHandler::flush() {
  std::shared_ptr<VPackBuilder> parsedRequest;
  VPackSlice slice;
  try {
    slice = _request->payload();
  } catch (...) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                  "invalid body value. expecting object");
    return;
  }
  if (!slice.isObject() && !slice.isNone()) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                  "invalid body value. expecting object");
  }
  
  bool waitForSync = false;
  bool waitForCollector = false;

  if (slice.isObject()) {
    // got a request body
    VPackSlice value = slice.get("waitForSync");
    if (value.isString()) {
      waitForSync = (value.copyString() == "true");
    } else if (value.isBoolean()) {
      waitForSync = value.getBoolean();
    }
    
    value = slice.get("waitForCollector");
    if (value.isString()) {
      waitForCollector = (value.copyString() == "true");
    } else if (value.isBoolean()) {
      waitForCollector = value.getBoolean();
    }
  } else {
    // no request body
    bool found;
    {
      std::string const& v = _request->value("waitForSync", found);
      if (found) {
        waitForSync = (v == "1" || v == "true");
      }
    }
    {
      std::string const& v = _request->value("waitForCollector", found);
      if (found) {
        waitForCollector = (v == "1" || v == "true");
      }
    }
  }
  
  int res;
  if (ServerState::instance()->isCoordinator()) {
    res = flushWalOnAllDBServers(waitForSync, waitForCollector);
  } else {
    res = arangodb::wal::LogfileManager::instance()->flush(
        waitForSync, waitForCollector, false);
  }

  if (res != TRI_ERROR_NO_ERROR) {
    THROW_ARANGO_EXCEPTION(res);
  }

  generateResult(rest::ResponseCode::OK, basics::VelocyPackHelper::EmptyObjectValue());
}

// Check all coordinators, guarded above doChecks
std::vector<check_t> Supervision::checkCoordinators() {
  std::vector<check_t> ret;
  Node::Children const& machinesPlanned =
      _snapshot(planCoordinatorsPrefix).children();
  Node::Children const serversRegistered =
      _snapshot(currentServersRegisteredPrefix).children();

  std::string currentFoxxmaster;
  try {
    currentFoxxmaster = _snapshot(foxxmaster).getString();
  } catch (...) {
  }

  std::string goodServerId;
  bool foxxmasterOk = false;
  std::vector<std::string> todelete;
  for (auto const& machine : _snapshot(healthPrefix).children()) {
    if (machine.first.substr(0, 2) == "Co") {
      todelete.push_back(machine.first);
    }
  }

  for (auto const& machine : machinesPlanned) {
    bool good = false;
    std::string lastHeartbeatTime, lastHeartbeatAcked, lastStatus,
        heartbeatTime, heartbeatStatus, serverID;

    serverID = machine.first;
    heartbeatTime = _snapshot(syncPrefix + serverID + "/time").toJson();
    heartbeatStatus = _snapshot(syncPrefix + serverID + "/status").toJson();

    todelete.erase(std::remove(todelete.begin(), todelete.end(), serverID),
                   todelete.end());

    try {  // Existing
      lastHeartbeatTime =
          _snapshot(healthPrefix + serverID + "/LastHeartbeatSent").toJson();
      lastStatus = _snapshot(healthPrefix + serverID + "/Status").toJson();
      if (lastHeartbeatTime != heartbeatTime) {  // Update
        good = true;
      }
    } catch (...) {  // New server
      good = true;
    }

    query_t report = std::make_shared<Builder>();
    report->openArray();
    report->openArray();
    report->openObject();
    report->add(_agencyPrefix + healthPrefix + serverID,
                VPackValue(VPackValueType::Object));
    report->add("LastHeartbeatSent", VPackValue(heartbeatTime));
    report->add("LastHeartbeatStatus", VPackValue(heartbeatStatus));
    report->add("Role", VPackValue("Coordinator"));
    auto endpoint = serversRegistered.find(serverID);
    if (endpoint != serversRegistered.end()) {
      endpoint = endpoint->second->children().find("endpoint");
      if (endpoint != endpoint->second->children().end()) {
        if (endpoint->second->children().size() == 0) {
          VPackSlice epString = endpoint->second->slice();
          if (epString.isString()) {
            report->add("Endpoint", epString);
          }
        }
      }
    }

    if (good) {
      if (goodServerId.empty()) {
        goodServerId = serverID;
      }
      if (serverID == currentFoxxmaster) {
        foxxmasterOk = true;
      }
      report->add(
          "LastHeartbeatAcked",
          VPackValue(timepointToString(std::chrono::system_clock::now())));
      report->add("Status", VPackValue(Supervision::HEALTH_STATUS_GOOD));
    } else {
      std::chrono::seconds t{0};
      t = std::chrono::duration_cast<std::chrono::seconds>(
          std::chrono::system_clock::now() -
          stringToTimepoint(lastHeartbeatAcked));
      if (t.count() > _gracePeriod) {  // Failure
        if (lastStatus == Supervision::HEALTH_STATUS_BAD) {
          report->add("Status", VPackValue(Supervision::HEALTH_STATUS_FAILED));
        }
      } else {
        report->add("Status", VPackValue(Supervision::HEALTH_STATUS_BAD));
      }
    }

    report->close();
    report->close();
    report->close();
    report->close();
    if (!this->isStopping()) {
      _agent->write(report);
    }
//.........这里部分代码省略.........

/// @brief local helper to throw an exception if a HTTP request went wrong
static bool throwExceptionAfterBadSyncRequest(ClusterCommResult* res,
                                              bool isShutdown) {
  DEBUG_BEGIN_BLOCK();
  if (res->status == CL_COMM_TIMEOUT ||
      res->status == CL_COMM_BACKEND_UNAVAILABLE) {
    THROW_ARANGO_EXCEPTION_MESSAGE(res->getErrorCode(),
                                   res->stringifyErrorMessage());
  }

  if (res->status == CL_COMM_ERROR) {
    std::string errorMessage;
    TRI_ASSERT(nullptr != res->result);

    arangodb::basics::StringBuffer const& responseBodyBuf(res->result->getBody());

    // extract error number and message from response
    int errorNum = TRI_ERROR_NO_ERROR;
    std::shared_ptr<VPackBuilder> builder = VPackParser::fromJson(
        responseBodyBuf.c_str(), responseBodyBuf.length());
    VPackSlice slice = builder->slice();

    if (!slice.hasKey("error") || slice.get("error").getBoolean()) {
      errorNum = TRI_ERROR_INTERNAL;
      errorMessage = std::string("Error message received from shard '") +
                     std::string(res->shardID) +
                     std::string("' on cluster node '") +
                     std::string(res->serverID) + std::st