arangodb · mchacki · Sep 9, 2021 · Jun 22, 2021 · Jun 23, 2021 · Jun 23, 2021
diff --git a/3rdParty/immer/v0.6.2/immer/transience/no_transience_policy.hpp b/3rdParty/immer/v0.6.2/immer/transience/no_transience_policy.hpp
@@ -26,7 +26,7 @@ struct no_transience_policy
             struct owner
             {
                 operator edit () const { return {}; }
-                owner& operator=(const owner&) { return *this; };
+                //owner& operator=(const owner&) { return *this; }; -- already fixed upstream
             };
 
             struct ownee

diff --git a/arangod/Agency/Supervision.cpp b/arangod/Agency/Supervision.cpp
@@ -2244,8 +2244,8 @@ void Supervision::checkReplicatedLogs() {
   };
 
   // check if Plan has replicated logs
-  auto const& node = snapshot().hasAsNode(planRepLogPrefix);
-  if (!node) {
+  auto const& planNode = snapshot().hasAsNode(planRepLogPrefix);
+  if (!planNode) {
     return;
   }
 
@@ -2267,20 +2267,29 @@ void Supervision::checkReplicatedLogs() {
 
   auto builder = std::make_shared<Builder>();
   auto envelope = arangodb::agency::envelope::into_builder(*builder);
-  for (auto const& [dbName, db] : node->get().children()) {
+  for (auto const& [dbName, db] : planNode->get().children()) {
     for (auto const& [idString, node] : db->children()) {
       auto spec = readPlanSpecification(*node);
-      auto current = std::invoke([&, &dbName = dbName, &idString = idString]() -> LogCurrent {
+      auto current = std::invoke([&, &dbName = dbName, &idString = idString]() -> std::optional<LogCurrent> {
         using namespace cluster::paths;
         auto currentPath =
             aliases::current()
                ->replicatedLogs()
                 ->database(dbName)
                 ->log(idString)
                 ->str(SkipComponents(1) /* skip first path component, i.e. 'arango' */);
-        return readLogCurrent(snapshot().get(currentPath)->get());
+
+        auto cnode = snapshot().get(currentPath);
+        if (cnode.has_value()) {
+          return readLogCurrent(cnode->get());
+        }
+        return std::nullopt;
       });
-      auto newTermSpec = checkReplicatedLog(dbName, spec, current, info);
+      if (!current.has_value()) {
+        continue;
+      }
+
+      auto newTermSpec = checkReplicatedLog(dbName, spec, *current, info);
 
       envelope = std::visit(
           overload{[&, &dbName = dbName](LogPlanTermSpecification const& newSpec) {

diff --git a/arangod/CMakeLists.txt b/arangod/CMakeLists.txt
@@ -566,7 +566,7 @@ set(LIB_ARANGO_REPLICATION2_SOURCES
   Replication2/ReplicatedLog/types.cpp
   Replication2/Version.cpp
   RestHandler/RestLogHandler.cpp
-  )
+)
 
 set (LIB_ARANGO_METRICS_SOURCES
   RestServer/Metrics.cpp

diff --git a/arangod/Replication2/AgencyMethods.cpp b/arangod/Replication2/AgencyMethods.cpp
@@ -20,13 +20,28 @@
 /// @author Lars Maier
 ////////////////////////////////////////////////////////////////////////////////
 
-#include <chrono>
-
 #include "AgencyMethods.h"
 
-#include <Agency/AsyncAgencyComm.h>
-#include <Agency/TransactionBuilder.h>
-#include <Cluster/ClusterFeature.h>
+#include <cstdint>
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include <velocypack/velocypack-aliases.h>
+#include <velocypack/velocypack-common.h>
+
+#include "Agency/AsyncAgencyComm.h"
+#include "Agency/TransactionBuilder.h"
+#include "Agency/AgencyPaths.h"
+#include "Cluster/ClusterTypes.h"
+#include "Replication2/ReplicatedLog/AgencyLogSpecification.h"
+#include "Replication2/ReplicatedLog/LogCommon.h"
+
+namespace arangodb {
+class Result;
+}  // namespace arangodb
 
 using namespace std::chrono_literals;
 

diff --git a/arangod/Replication2/AgencyMethods.h b/arangod/Replication2/AgencyMethods.h
@@ -21,13 +21,26 @@
 ////////////////////////////////////////////////////////////////////////////////
 
 #pragma once
-#include <optional>
-
 #include <Agency/TransactionBuilder.h>
 #include <Basics/ResultT.h>
 #include <Cluster/ClusterTypes.h>
-
 #include <Replication2/ReplicatedLog/AgencyLogSpecification.h>
+#include <optional>
+
+#include "Futures/Future.h"
+
+namespace arangodb {
+class Result;
+}  // namespace arangodb
+namespace arangodb::replication2 {
+class LogId;
+struct LogTerm;
+}  // namespace arangodb::replication2
+namespace arangodb::replication2::agency {
+struct LogCurrentSupervisionElection;
+struct LogPlanSpecification;
+struct LogPlanTermSpecification;
+}  // namespace arangodb::replication2::agency
 
 namespace arangodb::replication2::agency::methods {
 

diff --git a/arangod/Replication2/LoggerContext.h b/arangod/Replication2/LoggerContext.h
@@ -58,7 +58,7 @@ struct LogNameValuePair : LoggableValue {
 };
 
 struct LoggerContext {
-  explicit LoggerContext(LogTopic topic) : topic(std::move(topic)) {}
+  explicit LoggerContext(LogTopic const& topic) : topic(topic) {}
 
   template<const char N[], typename T>
   auto with(T&& t) const -> LoggerContext {
@@ -67,8 +67,8 @@ struct LoggerContext {
     return LoggerContext(values.push_back(std::move(pair)), topic);
   }
 
-  auto withTopic(LogTopic newTopic) const {
-    return LoggerContext(values, std::move(newTopic));
+  auto withTopic(LogTopic const& newTopic) const {
+    return LoggerContext(values, newTopic);
   }
 
   friend auto operator<<(std::ostream& os, LoggerContext const& ctx) -> std::ostream& {
@@ -85,12 +85,13 @@ struct LoggerContext {
     return os;
   }
 
-  LogTopic const topic;
-  ::immer::flex_vector<std::shared_ptr<LoggableValue>, arangodb::immer::arango_memory_policy> const values = {};
+  using Container = ::immer::flex_vector<std::shared_ptr<LoggableValue>, arangodb::immer::arango_memory_policy>;
+  LogTopic const& topic;
+  Container const values = {};
 
  private:
-  LoggerContext(decltype(values) values, LogTopic topic)
-      : topic(std::move(topic)), values(std::move(values)) {}
+  LoggerContext(Container values, LogTopic const& topic)
+      : topic(topic), values(std::move(values)) {}
 };
 }
 

diff --git a/arangod/Replication2/ReplicatedLog/AgencyLogSpecification.cpp b/arangod/Replication2/ReplicatedLog/AgencyLogSpecification.cpp
@@ -57,9 +57,9 @@ auto LogPlanTermSpecification::toVelocyPack(VPackBuilder& builder) const -> void
   }
 }
 
-LogPlanTermSpecification::LogPlanTermSpecification(from_velocypack_t, VPackSlice slice) {
-  term = slice.get(StaticStrings::Term).extract<LogTerm>();
-  config = LogConfig(slice.get(StaticStrings::Config));
+LogPlanTermSpecification::LogPlanTermSpecification(from_velocypack_t, VPackSlice slice)
+    : term(slice.get(StaticStrings::Term).extract<LogTerm>()),
+      config(slice.get(StaticStrings::Config)) {
   for (auto const& [key, value] :
        VPackObjectIterator(slice.get(StaticStrings::Participants))) {
     TRI_ASSERT(value.isEmptyObject());
@@ -82,9 +82,9 @@ auto LogPlanSpecification::toVelocyPack(VPackBuilder& builder) const -> void {
   }
 }
 
-LogPlanSpecification::LogPlanSpecification(from_velocypack_t, VPackSlice slice) {
-  id = slice.get(StaticStrings::Id).extract<LogId>();
-  targetConfig = LogConfig(slice.get(StaticStrings::TargetConfig));
+LogPlanSpecification::LogPlanSpecification(from_velocypack_t, VPackSlice slice)
+    : id(slice.get(StaticStrings::Id).extract<LogId>()),
+      targetConfig(slice.get(StaticStrings::TargetConfig)) {
   if (auto term = slice.get(StaticStrings::CurrentTerm); !term.isNone()) {
     currentTerm = LogPlanTermSpecification{from_velocypack, term};
   }
@@ -99,7 +99,8 @@ LogPlanTermSpecification::LogPlanTermSpecification(LogTerm term, LogConfig confi
       participants(std::move(participants)) {}
 
 LogPlanSpecification::LogPlanSpecification(LogId id, std::optional<LogPlanTermSpecification> term,
-                                           LogConfig config) : id(id), currentTerm(std::move(term)), targetConfig(config) {}
+                                           LogConfig config)
+    : id(id), currentTerm(std::move(term)), targetConfig(config) {}
 
 LogCurrentLocalState::LogCurrentLocalState(from_velocypack_t, VPackSlice slice) {
   auto spearheadSlice = slice.get(StaticStrings::Spearhead);
@@ -135,10 +136,11 @@ LogCurrentSupervision::LogCurrentSupervision(from_velocypack_t, VPackSlice slice
   }
 }
 
-LogCurrentSupervisionElection::LogCurrentSupervisionElection(from_velocypack_t, VPackSlice slice) {
-  term = slice.get(StaticStrings::Term).extract<LogTerm>();
-  participantsRequired = slice.get("participantsRequired").getNumericValue<std::size_t>();
-  participantsAvailable = slice.get("participantsAvailable").getNumericValue<std::size_t>();
+LogCurrentSupervisionElection::LogCurrentSupervisionElection(from_velocypack_t, VPackSlice slice)
+    : term(slice.get(StaticStrings::Term).extract<LogTerm>()),
+      participantsRequired(slice.get("participantsRequired").getNumericValue<std::size_t>()),
+      participantsAvailable(
+          slice.get("participantsAvailable").getNumericValue<std::size_t>()) {
   for (auto [key, value] : VPackObjectIterator(slice.get("details"))) {
     detail.emplace(key.copyString(), value.get("code").getNumericValue<ErrorCode>());
   }
@@ -172,21 +174,23 @@ auto LogCurrentSupervisionElection::toVelocyPack(VPackBuilder& builder) const ->
   builder.add("participantsAvailable", VPackValue(participantsAvailable));
   {
     VPackObjectBuilder db(&builder, "details");
-    for (auto const&[server, error] : detail) {
+    for (auto const& [server, error] : detail) {
       builder.add(VPackValue(server));
       ::toVelocyPack(error, builder);
     }
   }
 }
 
-auto agency::toVelocyPack(LogCurrentSupervisionElection::ErrorCode ec, VPackBuilder& builder) -> void {
+auto agency::toVelocyPack(LogCurrentSupervisionElection::ErrorCode ec,
+                          VPackBuilder& builder) -> void {
   VPackObjectBuilder ob(&builder);
   builder.add("code", VPackValue(static_cast<int>(ec)));
   builder.add("message", VPackValue(to_string(ec)));
 }
 
-auto agency::to_string(LogCurrentSupervisionElection::ErrorCode ec) noexcept -> std::string_view {
-  switch(ec) {
+auto agency::to_string(LogCurrentSupervisionElection::ErrorCode ec) noexcept
+    -> std::string_view {
+  switch (ec) {
     case LogCurrentSupervisionElection::ErrorCode::OK:
       return "the server is ok";
     case LogCurrentSupervisionElection::ErrorCode::SERVER_NOT_GOOD:

diff --git a/arangod/Replication2/ReplicatedLog/Algorithms.cpp b/arangod/Replication2/ReplicatedLog/Algorithms.cpp
@@ -223,8 +223,6 @@ auto algorithms::detectConflict(replicated_log::InMemoryLog const& log, TermInde
                             TermIndexPair{lastEntry->entry().logTerm(),
                                           lastEntry->entry().logIndex() + 1});
     } else {
-      // this can only happen if we drop log entries, check the code below before removing the assert
-      TRI_ASSERT(false);
       TRI_ASSERT(prevLog.index < lastEntry->entry().logIndex());
       TRI_ASSERT(prevLog.index < log.getFirstEntry()->entry().logIndex());
       // the given index too old, reset to (0, 0)

diff --git a/arangod/Replication2/ReplicatedLog/Algorithms.h b/arangod/Replication2/ReplicatedLog/Algorithms.h
@@ -27,8 +27,7 @@
 #include <unordered_map>

 #include "Cluster/ClusterTypes.h"
-#include "InMemoryLog.h"
-#include "ReplicatedLog.h"
+#include "Replication2/ReplicatedLog/ReplicatedLog.h"
 #include "Replication2/ReplicatedLog/AgencyLogSpecification.h"
 
 namespace arangodb::replication2::algorithms {

diff --git a/arangod/Replication2/ReplicatedLog/ILogParticipant.cpp b/arangod/Replication2/ReplicatedLog/ILogParticipant.cpp
@@ -28,6 +28,7 @@
 #include "RestServer/Metrics.h"
 
 #include <Basics/Exceptions.h>
+#include <Basics/StaticStrings.h>
 
 using namespace arangodb;
 using namespace arangodb::replication2;
@@ -64,3 +65,23 @@ auto replicated_log::ILogParticipant::waitForIterator(LogIndex index)
 auto replicated_log::ILogParticipant::getTerm() const noexcept -> std::optional<LogTerm> {
   return getStatus().getCurrentTerm();
 }
+
+auto replicated_log::LogUnconfiguredParticipant::release(LogIndex doneWithIdx) -> Result {
+  THROW_ARANGO_EXCEPTION(TRI_ERROR_NOT_IMPLEMENTED);
+}
+
+replicated_log::WaitForResult::WaitForResult(LogIndex index,
+                                             std::shared_ptr<QuorumData const> quorum)
+    : currentCommitIndex(index), quorum(std::move(quorum)) {}
+
+void replicated_log::WaitForResult::toVelocyPack(velocypack::Builder& builder) const {
+  VPackObjectBuilder ob(&builder);
+  builder.add(StaticStrings::CommitIndex, VPackValue(currentCommitIndex));
+  builder.add(VPackValue("quorum"));
+  quorum->toVelocyPack(builder);
+}
+
+replicated_log::WaitForResult::WaitForResult(velocypack::Slice s) {
+  currentCommitIndex = s.get(StaticStrings::CommitIndex).extract<LogIndex>();
+  quorum = std::make_shared<QuorumData>(s.get("quorum"));
+}
diff --git a/arangod/Replication2/ReplicatedLog/ILogParticipant.h b/arangod/Replication2/ReplicatedLog/ILogParticipant.h
@@ -33,11 +33,28 @@
 #include <map>
 #include <memory>
 
+namespace arangodb {
+class Result;
+}
+
 namespace arangodb::replication2::replicated_log {
 
 struct LogCore;
 struct LogStatus;
 
+struct WaitForResult {
+  /// @brief contains the _current_ commit index. (Not the index waited for)
+  LogIndex currentCommitIndex;
+  /// @brief Quorum information
+  std::shared_ptr<QuorumData const> quorum;
+
+  WaitForResult(LogIndex index, std::shared_ptr<QuorumData const> quorum);
+  WaitForResult() = default;
+  WaitForResult(velocypack::Slice);
+
+  void toVelocyPack(velocypack::Builder&) const;
+};
+
 /**
 * @brief Interface for a log participant: That is, usually either a leader or a
 * follower (LogLeader and LogFollower). Can also be a LogUnconfiguredParticipant,
@@ -51,21 +68,23 @@ struct ILogParticipant {
   [[nodiscard]] virtual auto resign() &&
       -> std::tuple<std::unique_ptr<LogCore>, DeferredAction> = 0;
 
-  using WaitForPromise = futures::Promise<std::shared_ptr<QuorumData const>>;
-  using WaitForFuture = futures::Future<std::shared_ptr<QuorumData const>>;
-  using WaitForIteratorFuture = futures::Future<std::unique_ptr<LogIterator>>;
+  using WaitForPromise = futures::Promise<WaitForResult>;
+  using WaitForFuture = futures::Future<WaitForResult>;
+  using WaitForIteratorFuture = futures::Future<std::unique_ptr<LogRangeIterator>>;
   using WaitForQueue = std::multimap<LogIndex, WaitForPromise>;
 
   [[nodiscard]] virtual auto waitFor(LogIndex index) -> WaitForFuture = 0;
   [[nodiscard]] virtual auto waitForIterator(LogIndex index) -> WaitForIteratorFuture;
   [[nodiscard]] virtual auto getTerm() const noexcept -> std::optional<LogTerm>;
+
+  [[nodiscard]] virtual auto release(LogIndex doneWithIdx) -> Result = 0;
 };
 
 /**
 * @brief Unconfigured log participant, i.e. currently neither a leader nor
 * follower. Holds a LogCore, does nothing else.
 */
-struct LogUnconfiguredParticipant
+struct LogUnconfiguredParticipant final
     : std::enable_shared_from_this<LogUnconfiguredParticipant>,
       ILogParticipant {
   ~LogUnconfiguredParticipant() override;
@@ -76,6 +95,7 @@ struct LogUnconfiguredParticipant
   auto resign() &&
       -> std::tuple<std::unique_ptr<LogCore>, DeferredAction> override;
   [[nodiscard]] auto waitFor(LogIndex) -> WaitForFuture override;
+  [[nodiscard]] auto release(LogIndex doneWithIdx) -> Result override;
  private:
   std::unique_ptr<LogCore> _logCore;
   std::shared_ptr<ReplicatedLogMetrics> const _logMetrics;