INDY-1199: Implement inconsistent 3PC state detection

plenum/config.py

@@ -310,3 +310,5 @@
 # 1 for recorder
 # 2 during replay
 STACK_COMPANION = 0
+
+ENABLE_NETWORK_I3PC_WATCHER = False

plenum/server/i3pc_watchers.py

@@ -0,0 +1,25 @@
+from typing import Callable, Iterable
+
+
+class NetworkI3PCWatcher:
+    def __init__(self, cb: Callable):
+        self.nodes = set()
+        self.connected = set()
+        self.callback = cb
+
+    def connect(self, name: str):
+        self.connected.add(name)
+
+    def disconnect(self, name: str):
+        had_consensus = self._has_consensus()
+        self.connected.discard(name)
+        if had_consensus and not self._has_consensus():
+            self.callback()
+
+    def set_nodes(self, nodes: Iterable[str]):
+        self.nodes = set(nodes)
+
+    def _has_consensus(self):
+        n = len(self.nodes)
+        f = (n - 1) // 3
+        return len(self.connected) > f

plenum/server/node.py

@@ -11,6 +11,7 @@
 
 from common.exceptions import LogicError
 from crypto.bls.bls_key_manager import LoadBLSKeyError
+from plenum.server.i3pc_watchers import NetworkI3PCWatcher
 from state.pruning_state import PruningState
 from state.state import State
 from storage.helper import initKeyValueStorage, initHashStore, initKeyValueStorageIntKeys
@@ -59,7 +60,7 @@
 from plenum.common.signer_simple import SimpleSigner
 from plenum.common.stacks import nodeStackClass, clientStackClass
 from plenum.common.startable import Status, Mode
-from plenum.common.txn_util import idr_from_req_data, get_from, get_req_id, \
+from plenum.common.txn_util import idr_from_req_data, get_req_id, \
     get_seq_no, get_type, get_payload_data, \
     get_txn_time, get_digest
 from plenum.common.types import PLUGIN_TYPE_VERIFICATION, \
@@ -246,8 +247,13 @@ def __init__(self,
         self.nodeInBox = deque()
         self.clientInBox = deque()
 
+        # 3PC state consistency watchdog based on network events
+        self.network_i3pc_watcher = NetworkI3PCWatcher(self.on_inconsistent_3pc_state_from_network)
+
         self.setPoolParams()
 
+        self.network_i3pc_watcher.connect(self.name)
+
         self.clientBlacklister = SimpleBlacklister(
             self.name + CLIENT_BLACKLISTER_SUFFIX)  # type: Blacklister
 
@@ -607,6 +613,14 @@ def on_view_change_complete(self):
             for replica in self.replicas:
                 replica.clear_requests_and_fix_last_ordered()
 
+    def on_inconsistent_3pc_state_from_network(self):
+        if self.config.ENABLE_NETWORK_I3PC_WATCHER:
+            self.on_inconsistent_3pc_state()
+
+    def on_inconsistent_3pc_state(self):
+        logger.warning("There is high probability that current 3PC state is inconsistent,"
+                       "immediate restart is recommended")
+
     def create_replicas(self) -> Replicas:
         return Replicas(self, self.monitor, self.config)
 
@@ -654,6 +668,7 @@ def loadSeqNoDB(self):
     def setPoolParams(self):
         # TODO should be always called when nodeReg is changed - automate
         self.allNodeNames = set(self.nodeReg.keys())
+        self.network_i3pc_watcher.set_nodes(self.allNodeNames)
         self.totalNodes = len(self.allNodeNames)
         self.f = getMaxFailures(self.totalNodes)
         self.requiredNumberOfInstances = self.f + 1  # per RBFT
@@ -1197,6 +1212,12 @@ def onConnsChanged(self, joined: Set[str], left: Set[str]):
             self.send_current_state_to_lagging_node(node)
             self.send_ledger_status_to_newly_connected_node(node)
 
+        for node in left:
+            self.network_i3pc_watcher.disconnect(node)
+
+        for node in joined:
+            self.network_i3pc_watcher.connect(node)
+
     def request_ledger_status_from_nodes(self, ledger_id, nodes=None):
         for node_name in nodes if nodes else self.nodeReg:
             if node_name == self.name:

plenum/test/restart/helper.py

@@ -20,7 +20,7 @@ def get_group(nodeSet, group_cnt, include_primary=False):
 
 
 def restart_nodes(looper, nodeSet, restart_set, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=None, restart_one_by_one=True):
+                  after_restart_timeout=None, start_one_by_one=True, wait_for_elections=True):
     for node_to_stop in restart_set:
         node_to_stop.cleanupOnStopping = True
         node_to_stop.stop()
@@ -42,10 +42,11 @@ def restart_nodes(looper, nodeSet, restart_set, tconf, tdir, allPluginsPath,
         idx = nodeSet.index(node_to_restart)
         nodeSet[idx] = restarted_node
         rest_nodes += [restarted_node]
-        if restart_one_by_one:
+        if start_one_by_one:
             looper.run(checkNodesConnected(rest_nodes))
 
-    if not restart_one_by_one:
+    if not start_one_by_one:
         looper.run(checkNodesConnected(nodeSet))
 
-    ensureElectionsDone(looper=looper, nodes=nodeSet)
+    if wait_for_elections:
+        ensureElectionsDone(looper=looper, nodes=nodeSet)

plenum/test/restart/test_network_i3pc_watcher.py

@@ -0,0 +1,101 @@
+import pytest
+
+from plenum.server.i3pc_watchers import NetworkI3PCWatcher
+
+DEFAULT_NODE_SET = {'Alpha', 'Beta', 'Gamma', 'Delta'}
+
+
+class WatcherCallbackMock:
+    def __init__(self):
+        self.call_count = 0
+
+    def __call__(self):
+        self.call_count += 1
+
+
+@pytest.fixture
+def watcher():
+    cb = WatcherCallbackMock()
+    watcher = NetworkI3PCWatcher(cb)
+    watcher.set_nodes(DEFAULT_NODE_SET)
+    return watcher
+
+
+def _add_node(watcher: NetworkI3PCWatcher, node: str):
+    nodes = watcher.nodes
+    nodes.add(node)
+    watcher.set_nodes(nodes)
+
+
+def _remove_node(watcher: NetworkI3PCWatcher, node: str):
+    nodes = watcher.nodes
+    nodes.discard(node)
+    watcher.set_nodes(nodes)
+
+
+def test_watcher_is_not_triggered_when_created(watcher: NetworkI3PCWatcher):
+    assert watcher.callback.call_count == 0
+
+
+def test_watcher_is_not_triggered_when_nodes_are_initially_connected(watcher: NetworkI3PCWatcher):
+    watcher.connect('Alpha')
+    watcher.connect('Beta')
+    watcher.connect('Gamma')
+    watcher.connect('Delta')
+
+    assert watcher.callback.call_count == 0
+
+
+def test_watcher_is_not_triggered_when_just_one_node_connects_and_disconnects(watcher: NetworkI3PCWatcher):
+    watcher.connect('Alpha')
+    watcher.disconnect('Alpha')
+
+    assert watcher.callback.call_count == 0
+
+
+def test_watcher_is_triggered_when_going_below_consensus(watcher: NetworkI3PCWatcher):
+    watcher.connect('Alpha')
+    watcher.connect('Beta')
+    watcher.disconnect('Beta')
+
+    assert watcher.callback.call_count == 1
+
+
+def test_watcher_is_not_triggered_when_adding_nodes_while_on_edge_of_consensus(watcher: NetworkI3PCWatcher):
+    watcher.connect('Alpha')
+    watcher.connect('Beta')
+    _add_node(watcher, 'Epsilon')
+    _add_node(watcher, 'Zeta')
+    _add_node(watcher, 'Eta')
+
+    assert watcher.callback.call_count == 0
+
+
+def test_watcher_is_not_triggered_when_removing_nodes_below_minimum_count(watcher: NetworkI3PCWatcher):
+    watcher.connect('Alpha')
+    watcher.connect('Beta')
+    watcher.connect('Gamma')
+    watcher.connect('Delta')
+    _remove_node(watcher, 'Delta')
+
+    assert watcher.callback.call_count == 0
+
+
+def test_watcher_is_not_triggered_when_removing_nodes_and_going_below_consensus(watcher: NetworkI3PCWatcher):
+    _add_node(watcher, 'Theta')
+    watcher.connect('Alpha')
+    watcher.connect('Theta')
+    _remove_node(watcher, 'Theta')
+
+    assert watcher.callback.call_count == 0
+
+
+def test_watcher_is_not_triggered_when_just_two_nodes_connect_and_disconnect_in_7_node_pool(
+        watcher: NetworkI3PCWatcher):
+    watcher.set_nodes(['Alpha', 'Beta', 'Gamma', 'Delta', 'Epsilon', 'Zeta', 'Eta'])
+    watcher.connect('Alpha')
+    watcher.connect('Beta')
+    watcher.disconnect('Alpha')
+    watcher.disconnect('Beta')
+
+    assert watcher.callback.call_count == 0

plenum/test/restart/test_restart_node_4_all.py

@@ -12,5 +12,5 @@ def test_restart_groups_4_of_7_np_no_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 4, include_primary=False)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=False)
+                  after_restart_timeout=tm, start_one_by_one=False)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes.py

@@ -12,5 +12,5 @@ def test_restart_groups_4_of_7_np_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 4, include_primary=False)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=True)
+                  after_restart_timeout=tm, start_one_by_one=True)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_4_all_wp.py

@@ -12,5 +12,5 @@ def test_restart_groups_4_of_7_wp_no_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 4, include_primary=True)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=False)
+                  after_restart_timeout=tm, start_one_by_one=False)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_4_np.py

@@ -12,5 +12,5 @@ def test_restart_groups_4_of_7_wp_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 4, include_primary=True)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=True)
+                  after_restart_timeout=tm, start_one_by_one=True)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_6_all_np.py

@@ -12,5 +12,5 @@ def test_restart_groups_6_of_7_np_no_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 6, include_primary=False)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=False)
+                  after_restart_timeout=tm, start_one_by_one=False)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_6_all_wp.py

@@ -12,5 +12,5 @@ def test_restart_groups_6_of_7_wp_no_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 6, include_primary=True)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=False)
+                  after_restart_timeout=tm, start_one_by_one=False)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_6_np.py

@@ -12,5 +12,5 @@ def test_restart_groups_6_of_7_np_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 6, include_primary=False)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=True)
+                  after_restart_timeout=tm, start_one_by_one=True)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_6_wp.py

@@ -12,5 +12,5 @@ def test_restart_groups_6_of_7_wp_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 6, include_primary=True)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=True)
+                  after_restart_timeout=tm, start_one_by_one=True)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_7.py

@@ -12,5 +12,5 @@ def test_restart_groups_7_of_7_wp_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 7, include_primary=False)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=True)
+                  after_restart_timeout=tm, start_one_by_one=True)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_nodes_7_all.py

@@ -12,5 +12,5 @@ def test_restart_groups_7_of_7_wp_no_tm(looper, txnPoolNodeSet, tconf, tdir,
     restart_group = get_group(txnPoolNodeSet, 7, include_primary=True)
 
     restart_nodes(looper, txnPoolNodeSet, restart_group, tconf, tdir, allPluginsPath,
-                  after_restart_timeout=tm, restart_one_by_one=False)
+                  after_restart_timeout=tm, start_one_by_one=False)
     sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/restart/test_restart_to_lower_view.py

@@ -0,0 +1,70 @@
+import pytest
+
+from plenum.test import waits
+from plenum.test.helper import sdk_send_random_and_check
+from plenum.test.node_request.helper import sdk_ensure_pool_functional
+from plenum.test.restart.helper import restart_nodes
+from plenum.test.view_change.helper import ensure_view_change_complete
+
+nodeCount = 4
+
+
+@pytest.fixture(scope="module")
+def tconf(tconf):
+    old_network_3pc_watcher_state = tconf.ENABLE_NETWORK_I3PC_WATCHER
+    tconf.ENABLE_NETWORK_I3PC_WATCHER = True
+    yield tconf
+    tconf.ENABLE_NETWORK_I3PC_WATCHER = old_network_3pc_watcher_state
+
+
+def test_restart_majority_to_lower_view(looper, txnPoolNodeSet, tconf, tdir, allPluginsPath,
+                                        sdk_pool_handle, sdk_wallet_client):
+    # Add transaction to ledger
+    sdk_send_random_and_check(looper, txnPoolNodeSet, sdk_pool_handle, sdk_wallet_client, 1)
+
+    # Move to higher view
+    ensure_view_change_complete(looper, txnPoolNodeSet)
+
+    majority = txnPoolNodeSet[:3]
+    minority = txnPoolNodeSet[3:]
+
+    # Restart majority group
+    tm = tconf.ToleratePrimaryDisconnection + waits.expectedPoolElectionTimeout(len(txnPoolNodeSet))
+    restart_nodes(looper, txnPoolNodeSet, majority, tconf, tdir, allPluginsPath,
+                  after_restart_timeout=tm, start_one_by_one=False, wait_for_elections=False)
+
+    # Check that nodes in minority group are aware that they might have inconsistent 3PC state
+    for node in minority:
+        assert node.spylog.count(node.on_inconsistent_3pc_state) == 1
+
+    # Check that nodes in majority group don't think they might have inconsistent 3PC state
+    for node in majority:
+        assert node.spylog.count(node.on_inconsistent_3pc_state) == 0
+
+    # Restart minority group
+    restart_nodes(looper, txnPoolNodeSet, minority, tconf, tdir, allPluginsPath,
+                  after_restart_timeout=tm, start_one_by_one=False)
+
+    # Check that all nodes are still functional
+    sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)
+
+
+def test_restart_half_to_lower_view(looper, txnPoolNodeSet, tconf, tdir, allPluginsPath,
+                                    sdk_pool_handle, sdk_wallet_client):
+    # Add transaction to ledger
+    sdk_send_random_and_check(looper, txnPoolNodeSet, sdk_pool_handle, sdk_wallet_client, 1)
+
+    # Move to higher view
+    ensure_view_change_complete(looper, txnPoolNodeSet)
+
+    # Restart half of nodes
+    tm = tconf.ToleratePrimaryDisconnection + waits.expectedPoolElectionTimeout(len(txnPoolNodeSet))
+    restart_nodes(looper, txnPoolNodeSet, txnPoolNodeSet[2:], tconf, tdir, allPluginsPath,
+                  after_restart_timeout=tm, start_one_by_one=False, wait_for_elections=False)
+
+    # Check that nodes don't think they may have inconsistent 3PC state
+    for node in txnPoolNodeSet:
+        assert node.spylog.count(node.on_inconsistent_3pc_state) == 0
+
+    # Check that all nodes are still functional
+    sdk_ensure_pool_functional(looper, txnPoolNodeSet, sdk_wallet_client, sdk_pool_handle)

plenum/test/test_node.py

@@ -350,7 +350,8 @@ def processRequest(self, request, frm):
                  Node.send_current_state_to_lagging_node,
                  Node.process_current_state_message,
                  Node.transmitToClient,
-                 Node.has_ordered_till_last_prepared_certificate
+                 Node.has_ordered_till_last_prepared_certificate,
+                 Node.on_inconsistent_3pc_state
                  ]