Sort out RestAgencyHandler behaviour for inquire.

arangodb · neunhoef · Mar 14, 2018 · Mar 7, 2018 · Mar 7, 2018 · Mar 8, 2018
commit c3a06a69acb70f0a5e5eede9e17df7acbbbffdc2
diff --git a/arangod/Agency/Agent.cpp b/arangod/Agency/Agent.cpp
@@ -255,6 +255,21 @@ AgentInterface::raft_commit_t Agent::waitFor(index_t index, double timeout) {
   return Agent::raft_commit_t::UNKNOWN;
 }
 
+// Check if log is committed up to index.
+bool Agent::isCommitted(index_t index) {
+
+  if (size() == 1) {  // single host agency
+    return true;
+  }
+
+  CONDITION_LOCKER(guard, _waitForCV);
+  if (leading()) {
+    return _commitIndex >= index;
+  } else {
+    return false;
+  }
+}
+
 //  AgentCallback reports id of follower and its highest processed index
 void Agent::reportIn(std::string const& peerId, index_t index, size_t toLog) {
 

diff --git a/arangod/Agency/Agent.h b/arangod/Agency/Agent.h
@@ -183,6 +183,9 @@ class Agent : public arangodb::Thread,
   /// @brief Wait for slaves to confirm appended entries
   AgentInterface::raft_commit_t waitFor(index_t last_entry, double timeout = 10.0) override;
 
+  /// @brief Check if everything up to a given index has been committed:
+  bool isCommitted(index_t last_entry) override;
+
   /// @brief Convencience size of agency
   size_t size() const;
 

diff --git a/arangod/Agency/AgentInterface.h b/arangod/Agency/AgentInterface.h
@@ -45,6 +45,9 @@ class AgentInterface {
   /// @brief Wait for slaves to confirm appended entries
   virtual raft_commit_t waitFor(index_t last_entry, double timeout = 2.0) = 0;
 
+  /// @brief Wait for slaves to confirm appended entries
+  virtual bool isCommitted(index_t last_entry) = 0;
+
   // Suffice warnings
   virtual ~AgentInterface() {};
 };

diff --git a/arangod/Agency/RestAgencyHandler.cpp b/arangod/Agency/RestAgencyHandler.cpp
@@ -287,9 +287,9 @@ RestStatus RestAgencyHandler::handleWrite() {
     body.close();
 
     if (result == Agent::raft_commit_t::UNKNOWN) {
-      generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
+      generateError(rest::ResponseCode::SERVICE_UNAVAILABLE, TRI_ERROR_HTTP_SERVICE_UNAVAILABLE);
     } else if (result == Agent::raft_commit_t::TIMEOUT) {
-      generateResult(rest::ResponseCode::REQUEST_TIMEOUT, body.slice());
+      generateError(rest::ResponseCode::REQUEST_TIMEOUT, 408);
     } else {
       if (errors > 0) { // Some/all requests failed
         generateResult(rest::ResponseCode::PRECONDITION_FAILED, body.slice());
@@ -402,78 +402,70 @@ RestStatus RestAgencyHandler::handleInquire() {
   }
 
   if (ret.accepted) {  // I am leading
-#if 0
-    // Stuff from write: needs adapting:
+
     bool found;
     std::string call_mode = _request->header("x-ara
10000
ngodb-agency-mode", found);
-    if (!found) { call_mode = "waitForCommitted"; }
-    size_t errors = 0;
-    Builder body;
-    body.openObject();
-    Agent::raft_commit_t result = Agent::raft_commit_t::OK;
+    if (!found) {
+      call_mode = "waitForCommitted";
+    }
 
-    if (call_mode != "noWait") {
-      // Note success/error
-      body.add("results", VPackValue(VPackValueType::Array));
-      for (auto const& index : ret.indices) {
-        body.add(VPackValue(index));
-        if (index == 0) {
-          errors++;
-        }
-      }
-      body.close();
+    // First possibility: The answer is empty, we have never heard about
+    // these transactions. In this case we say so, regardless what the
+    // "agency-mode" is.
+    // Second possibility: Non-empty answer, but agency-mode is "noWait",
+    // then we simply report our findings, too.
+    // Third possibility, we actually have a non-empty list of indices,
+    // and we need to wait for commit to answer.
 
-      // Wait for commit of highest except if it is 0?
-      if (!ret.indices.empty() && call_mode == "waitForCommitted") {
-        arangodb::consensus::index_t max_index = 0;
-        try {
-          max_index =
-            *std::max_element(ret.indices.begin(), ret.indices.end());
-        } catch (std::exception const& ex) {
-          LOG_TOPIC(WARN, Logger::AGENCY) << ex.what();
-        }
+    // Handle cases 2 and 3:
+    Agent::raft_commit_t result = Agent::raft_commit_t::OK;
+    bool allCommitted = true;
+    if (!ret.indices.empty()) {
+      arangodb::consensus::index_t max_index = 0;
+      try {
+        max_index =
+          *std::max_element(ret.indices.begin(), ret.indices.end());
+      } catch (std::exception const& ex) {
+        LOG_TOPIC(WARN, Logger::AGENCY) << ex.what();
+      }
 
-        if (max_index > 0) {
+      if (max_index > 0) {
+        if (call_mode == "waitForCommitted") {
           result = _agent->waitFor(max_index);
+        } else {
+          allCommitted = _agent->isCommitted(max_index);
         }
-
       }
     }
 
-    body.close();
+    // We can now prepare the result:
+    Builder body;
+    bool failed = false;
+    { VPackObjectBuilder b(&body);
+      body.add(VPackValue("results"));
+      { VPackArrayBuilder bb(&body);
+        for (auto const& index : ret.indices) {
+          body.add(VPackValue(index));
+          failed = (failed || index == 0);
+        }
+      }
+      body.add("inquired", VPackValue(true));
+      if (!allCommitted) {  // can only happen in agency_mode "noWait"
+        body.add("ongoing", VPackValue(true));
+      }
+    }
 
     if (result == Agent::raft_commit_t::UNKNOWN) {
-      generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
+      generateError(rest::ResponseCode::SERVICE_UNAVAILABLE, TRI_ERROR_HTTP_SERVICE_UNAVAILABLE);
     } else if (result == Agent::raft_commit_t::TIMEOUT) {
-      generateResult(rest::ResponseCode::REQUEST_TIMEOUT, body.slice());
+      generateError(rest::ResponseCode::REQUEST_TIMEOUT, 408);
     } else {
-      if (errors > 0) { // Some/all requests failed
+      if (failed > 0) { // Some/all requests failed
         generateResult(rest::ResponseCode::PRECONDITION_FAILED, body.slice());
-      } else {          // All good
+      } else {          // All good (or indeed unknown in case 1)
         generateResult(rest::ResponseCode::OK, body.slice());
       }
     }
-#endif
-
-    // The following is the old behaviour:
-    Builder body;
-    bool failed = false;
-    { VPackObjectBuilder b(&body);
-      if (ret.indices.empty()) {
-        body.add("ongoing", VPackValue(true));
-      } else {
-        body.add(VPackValue("results"));
-        { VPackArrayBuilder bb(&body);
-          for (auto const& index : ret.indices) {
-            body.add(VPackValue(index));
-            failed = (failed || index == 0);
-          }}
-
-      }
-      body.add("inquired", VPackValue(true));
-    }
-    generateResult(failed ? rest::ResponseCode::PRECONDITION_FAILED :
-                   rest::ResponseCode::OK, body.slice());
   } else {  // Redirect to leader
     if (_agent->leaderID() == NO_LEADER) {
       return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");

diff --git a/arangod/Cluster/ClusterInfo.cpp b/arangod/Cluster/ClusterInfo.cpp
@@ -1356,9 +1356,6 @@ int ClusterInfo::createCollectionCoordinator(std::string const& databaseName,
       }
 
     }
-
-    // Update our cache:
-    loadPlan();
   }
 
   if (numberOfShards == 0) {
-Original file line number
+Diff line change
@@ Expand Up @@
           }
         }
-        // Update our cache:
-        loadPlan();
       }
       if (numberOfShards == 0) {
@@ Expand Down @@