Cluster consistency: check for no clean store condition.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@884612 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 69 insertions, 26 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index fa53fc5475..f9ad734d79 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -260,6 +260,7 @@ Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
         store.load();
         if (store.getClusterId())
             clusterId = store.getClusterId(); // Use stored ID if there is one.
+        QPID_LOG(notice, "Cluster store state: " << store)
     }
 
     cpg.join(name);
diff --git a/cpp/src/qpid/cluster/InitialStatusMap.cpp b/cpp/src/qpid/cluster/InitialStatusMap.cpp
index a5618db3e6..59338f89d4 100644
--- a/cpp/src/qpid/cluster/InitialStatusMap.cpp
+++ b/cpp/src/qpid/cluster/InitialStatusMap.cpp
@@ -148,39 +148,49 @@ framing::Uuid InitialStatusMap::getClusterId() {
         return map.begin()->second->getClusterId(); // Youngest newcomer in node-id order
 }
 
+void checkId(Uuid& expect, const Uuid& actual, const string& msg) {
+    if (!expect) expect = actual;
+    assert(expect);
+    if (expect != actual)
+        throw Exception(msg);
+}
+
 void InitialStatusMap::checkConsistent() {
     assert(isComplete());
-    bool persistent = (map.begin()->second->getStoreState() != STORE_STATE_NO_STORE);
+    int clean = 0;
+    int dirty = 0;
+    int empty = 0;
+    int none = 0;
+    int active = 0;
     Uuid clusterId;
+    Uuid shutdownId;
+
     for (Map::iterator i = map.begin(); i != map.end(); ++i) {
-        // Must not mix transient and persistent members.
-        if (persistent != (i->second->getStoreState() != STORE_STATE_NO_STORE))
-            throw Exception("Mixing transient and persistent brokers in a cluster");
-        // Members with non-empty stores must have same cluster-id
+        if (i->second->getActive()) ++active;
         switch (i->second->getStoreState()) {
-          case STORE_STATE_NO_STORE:
-          case STORE_STATE_EMPTY_STORE:
-            break;               
+          case STORE_STATE_NO_STORE: ++none; break;
+          case STORE_STATE_EMPTY_STORE: ++empty; break;
           case STORE_STATE_DIRTY_STORE:
+            ++dirty;
+            checkId(clusterId, i->second->getClusterId(),
+                    "Cluster-ID mismatch. Stores belong to different clusters.");
+            break;
           case STORE_STATE_CLEAN_STORE:
-            if (!clusterId) clusterId = i->second->getClusterId();
-            assert(clusterId);
-            if (clusterId != i->second->getClusterId())
-                throw Exception("Cluster-id mismatch, brokers belonged to different clusters.");
-        }
-    }
-    // If this is a newly forming cluster, clean stores must have same shutdown-id
-    if (find_if(map.begin(), map.end(), &isActive) == map.end()) {
-        Uuid shutdownId;
-        for (Map::iterator i = map.begin(); i != map.end(); ++i) {
-            if (i->second->getStoreState() == STORE_STATE_CLEAN_STORE) {
-                if (!shutdownId) shutdownId = i->second->getShutdownId();
-                assert(shutdownId);
-                if (shutdownId != i->second->getShutdownId())
-                    throw Exception("Shutdown-id mismatch, brokers were not shut down together.");
-            }
+            ++clean;
+            checkId(clusterId, i->second->getClusterId(),
+                    "Cluster-ID mismatch. Stores belong to different clusters.");
+            checkId(shutdownId, i->second->getShutdownId(),
+                    "Shutdown-ID mismatch. Stores were not shut down together");
+            break;
         }
     }
+    // Can't mix transient and persistent members.
+    if (none && (clean+dirty+empty))
+        throw Exception("Mixing transient and persistent brokers in a cluster");
+    // If there are no active members and there are dirty stores there
+    // must be at least one clean store.
+    if (!active && dirty && !clean)
+        throw Exception("Cannot recover, no clean store");
 }
 
 
diff --git a/cpp/src/qpid/cluster/StoreStatus.cpp b/cpp/src/qpid/cluster/StoreStatus.cpp
index a7da3baa50..72e4891566 100644
--- a/cpp/src/qpid/cluster/StoreStatus.cpp
+++ b/cpp/src/qpid/cluster/StoreStatus.cpp
@@ -31,6 +31,7 @@ namespace cluster {
 using framing::Uuid;
 using namespace framing::cluster;
 using namespace boost::filesystem;
+using std::ostream;
 
 StoreStatus::StoreStatus(const std::string& d)
     : state(STORE_STATE_NO_STORE), dataDir(d)
@@ -90,5 +91,20 @@ void StoreStatus::clean(const Uuid& shutdownId_) {
     save();
 }
 
+ostream& operator<<(ostream& o, const StoreStatus& s) {
+    switch (s.getState()) {
+      case STORE_STATE_NO_STORE: o << "no store"; break;
+      case STORE_STATE_EMPTY_STORE: o << "empty store"; break;
+      case STORE_STATE_DIRTY_STORE:
+        o << "dirty store, cluster-id=" << s.getClusterId();
+        break;
+      case STORE_STATE_CLEAN_STORE:
+        o << "clean store, cluster-id=" << s.getClusterId()
+          << " shutdown-id=" << s.getShutdownId();
+        break;
+    }
+    return o;
+}
+
 }} // namespace qpid::cluster
 
diff --git a/cpp/src/qpid/cluster/StoreStatus.h b/cpp/src/qpid/cluster/StoreStatus.h
index 539f46c10b..522020ed69 100644
--- a/cpp/src/qpid/cluster/StoreStatus.h
+++ b/cpp/src/qpid/cluster/StoreStatus.h
@@ -24,6 +24,7 @@
 
 #include "qpid/framing/Uuid.h"
 #include "qpid/framing/enum.h"
+#include <iosfwd>
 
 namespace qpid {
 namespace cluster {
@@ -56,6 +57,8 @@ class StoreStatus
     Uuid clusterId, shutdownId;
     std::string dataDir;
 };
+
+std::ostream& operator<<(std::ostream&, const StoreStatus&);
 }} // namespace qpid::cluster
 
 #endif  /*!QPID_CLUSTER_STORESTATE_H*/
diff --git a/cpp/src/tests/cluster_tests.py b/cpp/src/tests/cluster_tests.py
index 78967196a9..e57a826000 100755
--- a/cpp/src/tests/cluster_tests.py
+++ b/cpp/src/tests/cluster_tests.py
@@ -18,7 +18,7 @@
 # under the License.
 #
 
-import os, signal, sys, time, imp
+import os, signal, sys, time, imp, re
 from qpid import datatypes, messaging
 from qpid.brokertest import *
 from qpid.harness import Skipped
@@ -189,4 +189,17 @@ class StoreTests(BrokerTest):
         a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False)
         b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=False)
 
-
+    def test_total_failure(self):
+        # Verify we abort with sutiable error message if no clean stores.
+        cluster = self.cluster(0, args=self.args()+["--cluster-size=2"])
+        a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False)
+        b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=True)
+        a.kill()
+        b.kill()
+        a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False)
+        b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=False)
+        assert a.wait() != 0
+        assert b.wait() != 0
+        msg = re.compile("critical.*no clean store")
+        assert msg.search(file(a.log).read())
+        assert msg.search(file(b.log).read())