<commit_before>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 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 <NAME>
////////////////////////////////////////////////////////////////////////////////

#include "ServerState.h"

#include <iomanip>
#include <sstream>

#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>

#include "Agency/AgencyComm.h"
#include "ApplicationFeatures/ApplicationServer.h"
#include "Basics/FileUtils.h"
#include "Basics/ReadLocker.h"
#include "Basics/VelocyPackHelper.h"
#include "Basics/WriteLocker.h"
#include "Cluster/ClusterInfo.h"

#include "Logger/Logger.h"
#include "RestServer/DatabaseFeature.h"
#include "RestServer/DatabasePathFeature.h"

using namespace arangodb;
using namespace arangodb::basics;

////////////////////////////////////////////////////////////////////////////////
/// @brief single instance of ServerState - will live as long as the server is
/// running
////////////////////////////////////////////////////////////////////////////////

static ServerState Instance;

ServerState::ServerState()
    : _id(),
      _dataPath(),
      _logPath(),
      _arangodPath(),
      _dbserverConfig(),
      _coordinatorConfig(),
      _address(),
      _lock(),
      _role(),
      _idOfPrimary(""),
      _state(STATE_UNDEFINED),
      _initialized(false),
      _clusterEnabled(false),
      _foxxmaster(""),
      _foxxmasterQueueupdate(false) {
  storeRole(ROLE_UNDEFINED);
}

ServerState::~ServerState() {}

////////////////////////////////////////////////////////////////////////////////
/// @brief create the (sole) instance
////////////////////////////////////////////////////////////////////////////////

ServerState* ServerState::instance() { return &Instance; }

////////////////////////////////////////////////////////////////////////////////
/// @brief get the string representation of a role
////////////////////////////////////////////////////////////////////////////////

const std::vector<std::string> ServerState::RoleStr ({
    "NONE", "SNGL", "PRMR", "SCND", "CRDN", "AGNT"
      });

const std::vector<std::string> ServerState::RoleStrReadable ({
    "none", "single", "dbserver", "secondary", "coordinator", "agent"
      });

std::string ServerState::roleToString(ServerState::RoleEnum role) {
  switch (role) {
    case ROLE_UNDEFINED:
      return "UNDEFINED";
    case ROLE_SINGLE:
      return "SINGLE";
    case ROLE_PRIMARY:
      return "PRIMARY";
    case ROLE_SECONDARY:
      return "SECONDARY";
    case ROLE_COORDINATOR:
      return "COORDINATOR";
    case ROLE_AGENT:
      return "AGENT";
  }

  TRI_ASSERT(false);
  return "";
}

////////////////////////////////////////////////////////////////////////////////
/// @brief convert a string to a role
////////////////////////////////////////////////////////////////////////////////

ServerState::RoleEnum ServerState::stringToRole(std::string const& value) {
  if (value == "SINGLE") {
    return ROLE_SINGLE;
  } else if (value == "PRIMARY") {
    return ROLE_PRIMARY;
  } else if (value == "SECONDARY") {
    return ROLE_SECONDARY;
  } else if (value == "COORDINATOR") {
    return ROLE_COORDINATOR;
  }

  return ROLE_UNDEFINED;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief convert a string representation to a state
////////////////////////////////////////////////////////////////////////////////

ServerState::StateEnum ServerState::stringToState(std::string const& value) {
  if (value == "SHUTDOWN") {
    return STATE_SHUTDOWN;
  }
  // TODO MAX: do we need to understand other states, too?

  return STATE_UNDEFINED;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief get the string representation of a state
////////////////////////////////////////////////////////////////////////////////

std::string ServerState::stateToString(StateEnum state) {
  // TODO MAX: cleanup
  switch (state) {
    case STATE_UNDEFINED:
      return "UNDEFINED";
    case STATE_STARTUP:
      return "STARTUP";
    case STATE_SERVINGASYNC:
      return "SERVING";
    case STATE_SERVINGSYNC:
      return "SERVING";
    case STATE_STOPPING:
      return "STOPPING";
    case STATE_STOPPED:
      return "STOPPED";
    case STATE_SYNCING:
      return "SYNCING";
    case STATE_INSYNC:
      return "INSYNC";
    case STATE_LOSTPRIMARY:
      return "LOSTPRIMARY";
    case STATE_SERVING:
      return "SERVING";
    case STATE_SHUTDOWN:
      return "SHUTDOWN";
  }

  TRI_ASSERT(false);
  return "";
}

////////////////////////////////////////////////////////////////////////////////
/// @brief find and set our role
////////////////////////////////////////////////////////////////////////////////
void ServerState::findAndSetRoleBlocking() {
  while (true) {
    auto role = determineRole(_localInfo, _id);
    std::string roleString = roleToString(role);
    LOG_TOPIC(DEBUG, Logger::CLUSTER) << "Found my role: " << roleString;

    if (storeRole(role)) {
      break;
    }
    sleep(1);
  }
}

////////////////////////////////////////////////////////////////////////////////
/// @brief flush the server state (used for testing)
////////////////////////////////////////////////////////////////////////////////

void ServerState::flush() { findAndSetRoleBlocking(); }

////////////////////////////////////////////////////////////////////////////////
/// @brief get the server role
////////////////////////////////////////////////////////////////////////////////

ServerState::RoleEnum ServerState::getRole() {
  auto role = loadRole();

  if (role != ServerState::ROLE_UNDEFINED || !_clusterEnabled) {
    return role;
  }

  //TRI_ASSERT(!_id.empty());
  findAndSetRoleBlocking();
  return loadRole();
}

bool ServerState::unregister() {
  TRI_ASSERT(!getId().empty());

  std::string const& id = getId();

  std::string localInfoEncoded = StringUtils::urlEncode(_localInfo);
  AgencyOperation deleteLocalIdMap("Target/MapLocalToID/" + localInfoEncoded,
                                   AgencySimpleOperationType::DELETE_OP);

  std::vector<AgencyOperation> operations = {deleteLocalIdMap};

  auto role = loadRole();
  const std::string agencyKey = roleToAgencyKey(role);
  TRI_ASSERT(isClusterRole(role));
  if (role == ROLE_COORDINATOR || role == ROLE_PRIMARY) {
    operations.push_back(AgencyOperation("Plan/" + agencyKey + "/" + id,
                                         AgencySimpleOperationType::DELETE_OP));
    operations.push_back(AgencyOperation("Current/" + agencyKey + "/" + id,
                                         AgencySimpleOperationType::DELETE_OP));
  }

  AgencyWriteTransaction unregisterTransaction(operations);

  AgencyComm comm;
  AgencyCommResult result;

  result = comm.sendTransactionWithFailover(unregisterTransaction);
  return result.successful();
}

bool ServerState::registerShortName(std::string const& id, ServerState::RoleEnum const& role) {
  // secondaries will not be handled here and will get assigned an auto generated one
  if (role == ROLE_SECONDARY) {
    return false;
  }
  AgencyComm comm;
  AgencyCommResult result;

  std::string agencyIdKey;
  std::string roleName;
  if (role == ROLE_PRIMARY) {
    agencyIdKey = "LatestDBServerId";
    roleName = "DBServer";
  } else {
    agencyIdKey = "LatestCoordinatorId";
    roleName = "Coordinator";
  }

  uint32_t shortNum(0);
  try {
    shortNum = std::stoul(id.substr(roleName.size(), 3));
  } catch (...) {
    LOG_TOPIC(DEBUG, Logger::CLUSTER) <<
      "Old id cannot be parsed for number.";
    return false;
  }

  const std::string idKey  = "Target/" + agencyIdKey;
  const std::string mapKey  = "Target/MapUniqueToShortID/" + id;
  
  size_t attempts {0};
  while (attempts++ < 300) {
    result = comm.getValues("Target/" + agencyIdKey);
    if (!result.successful()) {
      LOG_TOPIC(WARN, Logger::CLUSTER) << "Couldn't fetch Target/" + agencyIdKey;
      sleep(1);
      continue;
    }

    VPackSlice latestId = result.slice()[0].get(
    std::vector<std::string>(
      {AgencyCommManager::path(), "Target", agencyIdKey}));

    auto num = latestId.getNumber<uint32_t>();
    
    std::vector<AgencyOperation> operations;
    std::vector<AgencyPrecondition> preconditions;

    std::stringstream ss; // ShortName
    ss << roleName
      << std::setw(4) << std::setfill('0')
      << shortNum;
    std::string shortName = ss.str();

    VPackBuilder shortNumBuilder;
    shortNumBuilder.add(VPackValue(shortNum));

    VPackBuilder numBuilder;
    numBuilder.add(VPackValue(num));

    VPackBuilder mapBuilder;
    {
      VPackObjectBuilder b(&mapBuilder);
      mapBuilder.add("TransactionID", shortNumBuilder.slice());
      mapBuilder.add("ShortName", VPackValue(shortName));
    }

    operations.push_back(AgencyOperation(mapKey, AgencyValueOperationType::SET, mapBuilder.slice()));
    preconditions.push_back(
      AgencyPrecondition(idKey, AgencyPrecondition::Type::VALUE, numBuilder.slice())
    );
    if (num > shortNum) {
      // possible conflict! our shortname might already be taken!
      result = comm.getValues("Target/MapUniqueToShortID");

      if (!result.successful()) {
        LOG_TOPIC(WARN, Logger::CLUSTER) << "Couldn't fetch Target/MapUniqueToShortID";
        sleep(1);
        continue;
      }

      VPackSlice shortIdMap = result.slice()[0].get(std::vector<std::string>(
        {AgencyCommManager::path(), "Target", "MapUniqueToShortID"}));

      if (shortIdMap.isObject()) {
        for (auto const& s : VPackObjectIterator(shortIdMap)) {
          if (s.key.copyString() == "ShortName") {
            if (arangodb::basics::VelocyPackHelper::getStringValue(s.value, "") == shortName) {
              // our short name is taken. total disaster! very sad!
              return false;
            }
          }
        }
      }
    } else {
      // update the number so it the next auto generated number is out of our taken range
      operations.push_back({idKey, AgencyValueOperationType::SET, shortNumBuilder.slice()});
    }

    AgencyWriteTransaction trx(operations, preconditions);
    result = comm.sendTransactionWithFailover(trx);
    if (result.successful()) {
      return true;
    }
    sleep(1);
  }

  LOG_TOPIC(FATAL, Logger::STARTUP) << "Couldn't register shortname for " << id;
  FATAL_ERROR_EXIT();
  return false;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief try to integrate into a cluster
////////////////////////////////////////////////////////////////////////////////
bool ServerState::integrateIntoCluster(ServerState::RoleEnum role,
                                       std::string const& myAddress,
                                       std::string const& myId) {
  // id supplied via command line this is deprecated
  if (!myId.empty()) {
    if (!hasPersistedId()) {
      setId(myId);
      role = getRole();

      // we are known to the agency under our old id!
      if (role != ServerState::ROLE_UNDEFINED) {
        registerShortName(myId, role);
        writePersistedId(myId);
      } else {
        LOG_TOPIC(FATAL, Logger::STARTUP) << "started with --cluster.my-id but id unknown in agency!";
        FATAL_ERROR_EXIT();
      }
    } else {
      LOG_TOPIC(WARN, Logger::STARTUP) << "--cluster.my-id is deprecated and will be deleted.";
    }
  }

  AgencyComm comm;
  AgencyCommResult result;

  std::string id;
  if (!hasPersistedId()) {
    id = generatePersistedId(role);
    LOG_TOPIC(INFO, Logger::CLUSTER)
      << "Fresh start. Persisting new UUID " << id;
  } else {
    id = getPersistedId();
    LOG_TOPIC(DEBUG, Logger::CLUSTER)
      << "Restarting with persisted UUID " << id;
  }
  setId(id);

  if (!registerAtAgency(comm, role, id)) {
    FATAL_ERROR_EXIT();
  }

  const std::string agencyKey = roleToAgencyKey(role);
  const std::string planKey = "Plan/" + agencyKey + "/" + id;
  const std::string currentKey = "Current/" + agencyKey + "/" + id;

  _id = id;

  findAndSetRoleBlocking();
  LOG_TOPIC(DEBUG, Logger::CLUSTER) << "We successfully announced ourselves as "
    << roleToString(role) << " and our id is "
    << id;

  return true;
}

//////////////////////////////////////////////////////////////////////////////
/// @brief get the key for a role in the agency
//////////////////////////////////////////////////////////////////////////////
std::string ServerState::roleToAgencyKey(ServerState::RoleEnum role) {
  switch (role) {
    case ROLE_PRIMARY:
      return "DBServers";
    case ROLE_COORDINATOR:
      return "Coordinators";

    case ROLE_SECONDARY:<commit_msg>Fix unused variables<commit_after>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 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 <NAME>
////////////////////////////////////////////////////////////////////////////////

#include "ServerState.h"

#include <iomanip>
#include <sstream>

#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>

#include "Agency/AgencyComm.h"
#include "ApplicationFeatures/ApplicationServer.h"
#include "Basics/FileUtils.h"
#include "Basics/ReadLocker.h"
#include "Basics/VelocyPackHelper.h"
#include "Basics/WriteLocker.h"
#include "Cluster/ClusterInfo.h"

#include "Logger/Logger.h"
#include "RestServer/DatabaseFeature.h"
#include "RestServer/DatabasePathFeature.h"

using namespace arangodb;
using namespace arangodb::basics;

////////////////////////////////////////////////////////////////////////////////
/// @brief single instance of ServerState - will live as long as the server is
/// running
////////////////////////////////////////////////////////////////////////////////

static ServerState Instance;

ServerState::ServerState()
    : _id(),
      _dataPath(),
      _logPath(),
      _arangodPath(),
      _dbserverConfig(),
      _coordinatorConfig(),
      _address(),
      _lock(),
      _role(),
      _idOfPrimary(""),
      _state(STATE_UNDEFINED),
      _initialized(false),
      _clusterEnabled(false),
      _foxxmaster(""),
      _foxxmasterQueueupdate(false) {
  storeRole(ROLE_UNDEFINED);
}

ServerState::~ServerState() {}

////////////////////////////////////////////////////////////////////////////////
/// @brief create the (sole) instance
////////////////////////////////////////////////////////////////////////////////

ServerState* ServerState::instance() { return &Instance; }

////////////////////////////////////////////////////////////////////////////////
/// @brief get the string representation of a role
////////////////////////////////////////////////////////////////////////////////

const std::vector<std::string> ServerState::RoleStr ({
    "NONE", "SNGL", "PRMR", "SCND", "CRDN", "AGNT"
      });

const std::vector<std::string> ServerState::RoleStrReadable ({
    "none", "single", "dbserver", "secondary", "coordinator", "agent"
      });

std::string ServerState::roleToString(ServerState::RoleEnum role) {
  switch (role) {
    case ROLE_UNDEFINED:
      return "UNDEFINED";
    case ROLE_SINGLE:
      return "SINGLE";
    case ROLE_PRIMARY:
      return "PRIMARY";
    case ROLE_SECONDARY:
      return "SECONDARY";
    case ROLE_COORDINATOR:
      return "COORDINATOR";
    case ROLE_AGENT:
      return "AGENT";
  }

  TRI_ASSERT(false);
  return "";
}

////////////////////////////////////////////////////////////////////////////////
/// @brief convert a string to a role
////////////////////////////////////////////////////////////////////////////////

ServerState::RoleEnum ServerState::stringToRole(std::string const& value) {
  if (value == "SINGLE") {
    return ROLE_SINGLE;
  } else if (value == "PRIMARY") {
    return ROLE_PRIMARY;
  } else if (value == "SECONDARY") {
    return ROLE_SECONDARY;
  } else if (value == "COORDINATOR") {
    return ROLE_COORDINATOR;
  }

  return ROLE_UNDEFINED;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief convert a string representation to a state
////////////////////////////////////////////////////////////////////////////////

ServerState::StateEnum ServerState::stringToState(std::string const& value) {
  if (value == "SHUTDOWN") {
    return STATE_SHUTDOWN;
  }
  // TODO MAX: do we need to understand other states, too?

  return STATE_UNDEFINED;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief get the string representation of a state
////////////////////////////////////////////////////////////////////////////////

std::string ServerState::stateToString(StateEnum state) {
  // TODO MAX: cleanup
  switch (state) {
    case STATE_UNDEFINED:
      return "UNDEFINED";
    case STATE_STARTUP:
      return "STARTUP";
    case STATE_SERVINGASYNC:
      return "SERVING";
    case STATE_SERVINGSYNC:
      return "SERVING";
    case STATE_STOPPING:
      return "STOPPING";
    case STATE_STOPPED:
      return "STOPPED";
    case STATE_SYNCING:
      return "SYNCING";
    case STATE_INSYNC:
      return "INSYNC";
    case STATE_LOSTPRIMARY:
      return "LOSTPRIMARY";
    case STATE_SERVING:
      return "SERVING";
    case STATE_SHUTDOWN:
      return "SHUTDOWN";
  }

  TRI_ASSERT(false);
  return "";
}

////////////////////////////////////////////////////////////////////////////////
/// @brief find and set our role
////////////////////////////////////////////////////////////////////////////////
void ServerState::findAndSetRoleBlocking() {
  while (true) {
    auto role = determineRole(_localInfo, _id);
    std::string roleString = roleToString(role);
    LOG_TOPIC(DEBUG, Logger::CLUSTER) << "Found my role: " << roleString;

    if (storeRole(role)) {
      break;
    }
    sleep(1);
  }
}

////////////////////////////////////////////////////////////////////////////////
/// @brief flush the server state (used for testing)
////////////////////////////////////////////////////////////////////////////////

void ServerState::flush() { findAndSetRoleBlocking(); }

////////////////////////////////////////////////////////////////////////////////
/// @brief get the server role
////////////////////////////////////////////////////////////////////////////////

ServerState::RoleEnum ServerState::getRole() {
  auto role = loadRole();

  if (role != ServerState::ROLE_UNDEFINED || !_clusterEnabled) {
    return role;
  }

  //TRI_ASSERT(!_id.empty());
  findAndSetRoleBlocking();
  return loadRole();
}

bool ServerState::unregister() {
  TRI_ASSERT(!getId().empty());

  std::string const& id = getId();

  std::string localInfoEncoded = StringUtils::urlEncode(_localInfo);
  AgencyOperation deleteLocalIdMap("Target/MapLocalToID/" + localInfoEncoded,
                                   AgencySimpleOperationType::DELETE_OP);

  std::vector<AgencyOperation> operations = {deleteLocalIdMap};

  auto role = loadRole();
  const std::string agencyKey = roleToAgencyKey(role);
  TRI_ASSERT(isClusterRole(role));
  if (role == ROLE_COORDINATOR || role == ROLE_PRIMARY) {
    operations.push_back(AgencyOperation("Plan/" + agencyKey + "/" + id,
                                         AgencySimpleOperationType::DELETE_OP));
    operations.push_back(AgencyOperation("Current/" + agencyKey + "/" + id,
                                         AgencySimpleOperationType::DELETE_OP));
  }

  AgencyWriteTransaction unregisterTransaction(operations);

  AgencyComm comm;
  AgencyCommResult result;

  result = comm.sendTransactionWithFailover(unregisterTransaction);
  return result.successful();
}

bool ServerState::registerShortName(std::string const& id, ServerState::RoleEnum const& role) {
  // secondaries will not be handled here and will get assigned an auto generated one
  if (role == ROLE_SECONDARY) {
    return false;
  }
  AgencyComm comm;
  AgencyCommResult result;

  std::string agencyIdKey;
  std::string roleName;
  if (role == ROLE_PRIMARY) {
    agencyIdKey = "LatestDBServerId";
    roleName = "DBServer";
  } else {
    agencyIdKey = "LatestCoordinatorId";
    roleName = "Coordinator";
  }

  uint32_t shortNum(0);
  try {
    shortNum = std::stoul(id.substr(roleName.size(), 3));
  } catch (...) {
    LOG_TOPIC(DEBUG, Logger::CLUSTER) <<
      "Old id cannot be parsed for number.";
    return false;
  }

  const std::string idKey  = "Target/" + agencyIdKey;
  const std::string mapKey  = "Target/MapUniqueToShortID/" + id;
  
  size_t attempts {0};
  while (attempts++ < 300) {
    result = comm.getValues("Target/" + agencyIdKey);
    if (!result.successful()) {
      LOG_TOPIC(WARN, Logger::CLUSTER) << "Couldn't fetch Target/" + agencyIdKey;
      sleep(1);
      continue;
    }

    VPackSlice latestId = result.slice()[0].get(
    std::vector<std::string>(
      {AgencyCommManager::path(), "Target", agencyIdKey}));

    auto num = latestId.getNumber<uint32_t>();
    
    std::vector<AgencyOperation> operations;
    std::vector<AgencyPrecondition> preconditions;

    std::stringstream ss; // ShortName
    ss << roleName
      << std::setw(4) << std::setfill('0')
      << shortNum;
    std::string shortName = ss.str();

    VPackBuilder shortNumBuilder;
    shortNumBuilder.add(VPackValue(shortNum));

    VPackBuilder numBuilder;
    numBuilder.add(VPackValue(num));

    VPackBuilder mapBuilder;
    {
      VPackObjectBuilder b(&mapBuilder);
      mapBuilder.add("TransactionID", shortNumBuilder.slice());
      mapBuilder.add("ShortName", VPackValue(shortName));
    }

    operations.push_back(AgencyOperation(mapKey, AgencyValueOperationType::SET, mapBuilder.slice()));
    preconditions.push_back(
      AgencyPrecondition(idKey, AgencyPrecondition::Type::VALUE, numBuilder.slice())
    );
    if (num > shortNum) {
      // possible conflict! our shortname might already be taken!
      result = comm.getValues("Target/MapUniqueToShortID");

      if (!result.successful()) {
        LOG_TOPIC(WARN, Logger::CLUSTER) << "Couldn't fetch Target/MapUniqueToShortID";
        sleep(1);
        continue;
      }

      VPackSlice shortIdMap = result.slice()[0].get(std::vector<std::string>(
        {AgencyCommManager::path(), "Target", "MapUniqueToShortID"}));

      if (shortIdMap.isObject()) {
        for (auto const& s : VPackObjectIterator(shortIdMap)) {
          if (s.key.copyString() == "ShortName") {
            if (arangodb::basics::VelocyPackHelper::getStringValue(s.value, "") == shortName) {
              // our short name is taken. total disaster! very sad!
              return false;
            }
          }
        }
      }
    } else {
      // update the number so it the next auto generated number is out of our taken range
      operations.push_back({idKey, AgencyValueOperationType::SET, shortNumBuilder.slice()});
    }

    AgencyWriteTransaction trx(operations, preconditions);
    result = comm.sendTransactionWithFailover(trx);
    if (result.successful()) {
      return true;
    }
    sleep(1);
  }

  LOG_TOPIC(FATAL, Logger::STARTUP) << "Couldn't register shortname for " << id;
  FATAL_ERROR_EXIT();
  return false;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief try to integrate into a cluster
////////////////////////////////////////////////////////////////////////////////
bool ServerState::integrateIntoCluster(ServerState::RoleEnum role,
                                       std::string const& myAddress,
                                       std::string const& myId) {
  // id supplied via command line this is deprecated
  if (!myId.empty()) {
    if (!hasPersistedId()) {
      setId(myId);
      role = getRole();

      // we are known to the agency under our old id!
      if (role != ServerState::ROLE_UNDEFINED) {
        registerShortName(myId, role);
        writePersistedId(myId);
      } else {
        LOG_TOPIC(FATAL, Logger::STARTUP) << "started with --cluster.my-id but id unknown in agency!";
        FATAL_ERROR_EXIT();
      }
    } else {
      LOG_TOPIC(WARN, Logger::STARTUP) << "--cluster.my-id is deprecated and will be deleted.";
    }
  }

  AgencyComm comm;
  AgencyCommResult result;

  std::string id;
  if (!hasPersistedId()) {
    id = generatePersistedId(role);
    LOG_TOPIC(INFO, Logger::CLUSTER)
      << "Fresh start. Persisting new UUID " << id;
  } else {
    id = getPersistedId();
    LOG_TOPIC(DEBUG, Logger::CLUSTER)
      << "Restarting with persisted UUID " << id;
  }
  setId(id);

  if (!registerAtAgency(comm, role, id)) {
    FATAL_ERROR_EXIT();
  }

  findAndSetRoleBlocking();
  LOG_TOPIC(DEBUG, Logger::CLUSTER) << "We successfully announced ourselves as "
    << roleToString(role) << " and our id is "
    << id;

  return true;
}

//////////////////////////////////////////////////////////////////////////////
/// @brief get the key for a role in the agency
//////////////////////////////////////////////////////////////////////////////
std::string ServerState::roleToAgencyKey(ServerState::RoleEnum role) {
  switch (role) {
    case ROLE_PRIMARY:
      return "DBServers";
    case ROLE_COORDINATOR:
      return "Coordinators";

    case ROLE_SECONDARY:
{'fraction_non_alphanumeric': 0.2536280923377698, 'fraction_numerical': 0.0028573576960673735, 'mean_word_length': 1.8971789565406818, 'pattern_counts': {'":': 0, '<': 117, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1858554', 'n_tokens_mistral': 7295, 'n_tokens_neox': 7132, 'n_words': 1717}