<commit_before>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2019 ArangoDB 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 "ClusterTrxMethods.h"

#include "Basics/NumberUtils.h"
#include "Basics/StaticStrings.h"
#include "Basics/StringUtils.h"
#include "Basics/VelocyPackHelper.h"
#include "Cluster/ClusterComm.h"
#include "Cluster/ClusterInfo.h"
#include "Cluster/ClusterMethods.h"
#include "Cluster/FollowerInfo.h"
#include "StorageEngine/TransactionCollection.h"
#include "StorageEngine/TransactionState.h"
#include "Transaction/Context.h"
#include "Transaction/Helpers.h"
#include "Transaction/Methods.h"
#include "Utils/OperationOptions.h"
#include "VocBase/LogicalCollection.h"

#include <velocypack/Buffer.h>
#include <velocypack/Iterator.h>
#include <velocypack/Slice.h>
#include <velocypack/velocypack-aliases.h>

using namespace arangodb;
using namespace arangodb::basics;

namespace {

// Timeout for read operations:
static double const CL_DEFAULT_TIMEOUT = 120.0;

void buildTransactionBody(TransactionState& state, ServerID const& server,
                          VPackBuilder& builder) {
  // std::vector<ServerID> DBservers = ci->getCurrentDBServers();
  builder.openObject();
  state.options().toVelocyPack(builder);
  builder.add("collections", VPackValue(VPackValueType::Object));
  auto addCollections = [&](std::string const& key, AccessMode::Type t) {
    builder.add(key, VPackValue(VPackValueType::Array));
    state.allCollections([&](TransactionCollection& col) {
      if (col.accessType() != t) {
        return true;
      }
      if (!state.isCoordinator()) {
        if (col.collection()->followers()->contains(server)) {
          builder.add(VPackValue(col.collectionName()));
        }
        return true;
      }
      
      // coordinator starts transaction on shard leaders
#ifdef USE_ENTERPRISE
      if (col.collection()->isSmart() && col.collection()->type() == TRI_COL_TYPE_EDGE) {
        auto names = col.collection()->realNames();
        auto* ci = ClusterInfo::instance();
        for (std::string const& name : names) {
          auto cc = ci->getCollectionNT(state.vocbase().name(), name);
          if (!cc) {
            continue;
          }
          auto shards = ci->getShardList(std::to_string(cc->id()));
          for (ShardID const& shard : *shards) {
            auto sss = ci->getResponsibleServer(shard);
            if (server == sss->at(0)) {
              builder.add(VPackValue(shard));
            }
          }
        }
        return true;
      }
#endif
      std::shared_ptr<ShardMap> shardIds = col.collection()->shardIds();
      for (auto const& pair : *shardIds) {
        TRI_ASSERT(!pair.second.empty());
        // only add shard where server is leader
        if (!pair.second.empty() && pair.second[0] == server) {
          builder.add(VPackValue(pair.first));
        }
      }
      return true;
    });
    builder.close();
  };
  addCollections("read", AccessMode::Type::READ);
  addCollections("write", AccessMode::Type::WRITE);
  addCollections("exclusive", AccessMode::Type::EXCLUSIVE);

  builder.close();  // </collections>
  builder.close();  // </openObject>
}

/// @brief lazy begin a transaction on subordinate servers
ClusterCommRequest beginTransactionRequest(transaction::Methods const* trx,
                                           TransactionState& state, ServerID const& server) {
  TRI_voc_tid_t tid = state.id() + 1;
  TRI_ASSERT(!transaction::isLegacyTransactionId(tid));

  VPackBuilder builder;
  buildTransactionBody(state, server, builder);

  const std::string url = "/_db/" + StringUtils::urlEncode(state.vocbase().name()) +
                          "/_api/transaction/begin";
  auto headers = std::make_unique<std::unordered_map<std::string, std::string>>();
  headers->emplace(StaticStrings::ContentTypeHeader, StaticStrings::MimeTypeJson);
  headers->emplace(StaticStrings::TransactionId, std::to_string(tid));
  auto body = std::make_shared<std::string>(builder.slice().toJson());
  return ClusterCommRequest("server:" + server, RequestType::POST, url, body,
                            std::move(headers));
}

/// check the transaction cluster response with desited TID and status
Result checkTransactionResult(TRI_voc_tid_t desiredTid,
                              transaction::Status desStatus,
                              ClusterCommResult const& result) {
  Result res;

  int commError = ::handleGeneralCommErrors(&result);
  if (commError != TRI_ERROR_NO_ERROR) {
    // oh-oh cluster is in a bad state
    return res.reset(commError);
  }
  TRI_ASSERT(result.status == CL_COMM_RECEIVED);

  VPackSlice answer = result.answer->payload();
  if ((result.answer_code == rest::ResponseCode::OK ||
       result.answer_code == rest::ResponseCode::CREATED) &&
      answer.isObject()) {
    VPackSlice idSlice = answer.get(std::vector<std::string>{"result", "id"});
    VPackSlice statusSlice =
        answer.get(std::vector<std::string>{"result", "status"});

    if (!idSlice.isString() || !statusSlice.isString()) {
      return res.reset(TRI_ERROR_TRANSACTION_INTERNAL,
                          "transaction has wrong format");
    }
    TRI_voc_tid_t tid = StringUtils::uint64(idSlice.copyString());
    VPackValueLength len = 0;
    const char* str = statusSlice.getStringUnchecked(len);
    if (tid == desiredTid && transaction::statusFromString(str, len) == desStatus) {
      return res;  // success
    }
  } else if (answer.isObject()) {
    std::string msg = std::string("error while ");
    if (desStatus == transaction::Status::RUNNING) {
      msg.append("beginning transaction");
    } else if (desStatus == transaction::Status::COMMITTED) {
      msg.append("committing transaction");
    } else if (desStatus == transaction::Status::RUNNING) {
      msg.append("aborting transaction");
    }                      
    return res.reset(VelocyPackHelper::readNumericValue(answer, StaticStrings::ErrorNum,
                                                        TRI_ERROR_TRANSACTION_INTERNAL),
                        VelocyPackHelper::getStringValue(answer, StaticStrings::ErrorMessage,
                                                         msg));
  }
  LOG_TOPIC(DEBUG, Logger::TRANSACTIONS)
      << " failed to begin transaction on " << result.endpoint;<commit_msg>fix compile warning<commit_after>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2019 ArangoDB 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 "ClusterTrxMethods.h"

#include "Basics/NumberUtils.h"
#include "Basics/StaticStrings.h"
#include "Basics/StringUtils.h"
#include "Basics/VelocyPackHelper.h"
#include "Cluster/ClusterComm.h"
#include "Cluster/ClusterInfo.h"
#include "Cluster/ClusterMethods.h"
#include "Cluster/FollowerInfo.h"
#include "StorageEngine/TransactionCollection.h"
#include "StorageEngine/TransactionState.h"
#include "Transaction/Context.h"
#include "Transaction/Helpers.h"
#include "Transaction/Methods.h"
#include "Utils/OperationOptions.h"
#include "VocBase/LogicalCollection.h"

#include <velocypack/Buffer.h>
#include <velocypack/Iterator.h>
#include <velocypack/Slice.h>
#include <velocypack/velocypack-aliases.h>

using namespace arangodb;
using namespace arangodb::basics;

namespace {

// Timeout for read operations:
static double const CL_DEFAULT_TIMEOUT = 120.0;

void buildTransactionBody(TransactionState& state, ServerID const& server,
                          VPackBuilder& builder) {
  // std::vector<ServerID> DBservers = ci->getCurrentDBServers();
  builder.openObject();
  state.options().toVelocyPack(builder);
  builder.add("collections", VPackValue(VPackValueType::Object));
  auto addCollections = [&](std::string const& key, AccessMode::Type t) {
    builder.add(key, VPackValue(VPackValueType::Array));
    state.allCollections([&](TransactionCollection& col) {
      if (col.accessType() != t) {
        return true;
      }
      if (!state.isCoordinator()) {
        if (col.collection()->followers()->contains(server)) {
          builder.add(VPackValue(col.collectionName()));
        }
        return true;
      }
      
      // coordinator starts transaction on shard leaders
#ifdef USE_ENTERPRISE
      if (col.collection()->isSmart() && col.collection()->type() == TRI_COL_TYPE_EDGE) {
        auto names = col.collection()->realNames();
        auto* ci = ClusterInfo::instance();
        for (std::string const& name : names) {
          auto cc = ci->getCollectionNT(state.vocbase().name(), name);
          if (!cc) {
            continue;
          }
          auto shards = ci->getShardList(std::to_string(cc->id()));
          for (ShardID const& shard : *shards) {
            auto sss = ci->getResponsibleServer(shard);
            if (server == sss->at(0)) {
              builder.add(VPackValue(shard));
            }
          }
        }
        return true;
      }
#endif
      std::shared_ptr<ShardMap> shardIds = col.collection()->shardIds();
      for (auto const& pair : *shardIds) {
        TRI_ASSERT(!pair.second.empty());
        // only add shard where server is leader
        if (!pair.second.empty() && pair.second[0] == server) {
          builder.add(VPackValue(pair.first));
        }
      }
      return true;
    });
    builder.close();
  };
  addCollections("read", AccessMode::Type::READ);
  addCollections("write", AccessMode::Type::WRITE);
  addCollections("exclusive", AccessMode::Type::EXCLUSIVE);

  builder.close();  // </collections>
  builder.close();  // </openObject>
}

/// @brief lazy begin a transaction on subordinate servers
ClusterCommRequest beginTransactionRequest(transaction::Methods const* trx,
                                           TransactionState& state, ServerID const& server) {
  TRI_voc_tid_t tid = state.id() + 1;
  TRI_ASSERT(!transaction::isLegacyTransactionId(tid));

  VPackBuilder builder;
  buildTransactionBody(state, server, builder);

  const std::string url = "/_db/" + StringUtils::urlEncode(state.vocbase().name()) +
                          "/_api/transaction/begin";
  auto headers = std::make_unique<std::unordered_map<std::string, std::string>>();
  headers->emplace(StaticStrings::ContentTypeHeader, StaticStrings::MimeTypeJson);
  headers->emplace(StaticStrings::TransactionId, std::to_string(tid));
  auto body = std::make_shared<std::string>(builder.slice().toJson());
  return ClusterCommRequest("server:" + server, RequestType::POST, url, body,
                            std::move(headers));
}

/// check the transaction cluster response with desited TID and status
Result checkTransactionResult(TRI_voc_tid_t desiredTid,
                              transaction::Status desStatus,
                              ClusterCommResult const& result) {
  Result res;

  int commError = ::handleGeneralCommErrors(&result);
  if (commError != TRI_ERROR_NO_ERROR) {
    // oh-oh cluster is in a bad state
    return res.reset(commError);
  }
  TRI_ASSERT(result.status == CL_COMM_RECEIVED);

  VPackSlice answer = result.answer->payload();
  if ((result.answer_code == rest::ResponseCode::OK ||
       result.answer_code == rest::ResponseCode::CREATED) &&
      answer.isObject()) {
    VPackSlice idSlice = answer.get(std::vector<std::string>{"result", "id"});
    VPackSlice statusSlice =
        answer.get(std::vector<std::string>{"result", "status"});

    if (!idSlice.isString() || !statusSlice.isString()) {
      return res.reset(TRI_ERROR_TRANSACTION_INTERNAL,
                          "transaction has wrong format");
    }
    TRI_voc_tid_t tid = StringUtils::uint64(idSlice.copyString());
    VPackValueLength len = 0;
    const char* str = statusSlice.getStringUnchecked(len);
    if (tid == desiredTid && transaction::statusFromString(str, len) == desStatus) {
      return res;  // success
    }
  } else if (answer.isObject()) {
    std::string msg = std::string("error while ");
    if (desStatus == transaction::Status::RUNNING) {
      msg.append("beginning transaction");
    } else if (desStatus == transaction::Status::COMMITTED) {
      msg.append("committing transaction");
    } else if (desStatus == transaction::Status::ABORTED) {
      msg.append("aborting transaction");
    }                      
    return res.reset(VelocyPackHelper::readNumericValue(answer, StaticStrings::ErrorNum,
                                                        TRI_ERROR_TRANSACTION_INTERNAL),
                        VelocyPackHelper::getStringValue(answer, StaticStrings::ErrorMessage,
                                                         msg));
  }
  LOG_TOPIC(DEBUG, Logger::TRANSACTIONS)
      << " failed to begin transaction on " << result.endpoint;
{'fraction_non_alphanumeric': 0.15458257074475573, 'fraction_numerical': 0.002534140504012389, 'mean_word_length': 2.8386922453390975, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3815360', 'n_tokens_mistral': 4008, 'n_tokens_neox': 3923, 'n_words': 887}