Merge pull request #819 from spivachuk/forced-upgrade-fix

INDY-1447: Fixed handling of forced requests

Author: ashcherbakov

plenum/persistence/req_id_to_txn.py

@@ -1,5 +1,4 @@
 import string
-from typing import Optional
 
 from storage.kv_store import KeyValueStorage
 
@@ -14,13 +13,13 @@ class ReqIdrToTxn:
     def __init__(self, keyValueStorage: KeyValueStorage):
         self._keyValueStorage = keyValueStorage
 
-    def add(self, digest, ledge_id, seq_no):
-        self._keyValueStorage.put(digest, self._create_value(ledge_id, seq_no))
+    def add(self, digest, ledger_id, seq_no):
+        self._keyValueStorage.put(digest, self._create_value(ledger_id, seq_no))
 
     def addBatch(self, batch):
-        self._keyValueStorage.setBatch([(digest, self._create_value(ledge_id,
+        self._keyValueStorage.setBatch([(digest, self._create_value(ledger_id,
                                                                     seq_no))
-                                        for digest, ledge_id, seq_no in batch])
+                                        for digest, ledger_id, seq_no in batch])
 
     def get(self, digest):
         """
@@ -37,7 +36,7 @@ def get(self, digest):
 
     def _parse_value(self, val: string):
         parse_data = val.split(self.delimiter)
-        return str(parse_data[0]), int(parse_data[1])
+        return int(parse_data[0]), int(parse_data[1])
 
     def _create_value(self, ledger_id, seq_no):
         return str(ledger_id) + self.delimiter + str(seq_no)

plenum/server/ledger_req_handler.py

@@ -1,7 +1,7 @@
 from abc import ABCMeta, abstractmethod
 from typing import List
 
-from common.exceptions import PlenumValueError
+from common.exceptions import PlenumValueError, LogicError
 from common.serializers.serialization import state_roots_serializer
 from stp_core.common.log import getlogger
 
@@ -66,6 +66,11 @@ def commit(self, txnCount, stateRoot, txnRoot, ppTime) -> List:
         return self._commit(self.ledger, self.state, txnCount, stateRoot,
                             txnRoot, ppTime, ts_store=self.ts_store)
 
+    def applyForced(self, req: Request):
+        if not req.isForced():
+            raise LogicError('requestHandler.applyForce method is called '
+                             'for not forced request: {}'.format(req))
+
     def onBatchCreated(self, state_root):
         pass
 

plenum/server/node.py

@@ -2130,32 +2130,46 @@ def processRequest(self, request: Request, frm: str):
         # TODO: What if the reply was a REQNACK? Its not gonna be found in the
         # replies.
 
-        if request.operation[TXN_TYPE] == GET_TXN:
+        txn_type = request.operation[TXN_TYPE]
+
+        if self.is_action(txn_type):
+            self.process_action(request, frm)
+
+        elif txn_type == GET_TXN:
             self.handle_get_txn_req(request, frm)
             self.total_read_request_number += 1
-            return
-
-        ledger_id = self.ledger_id_for_request(request)
-        ledger = self.getLedger(ledger_id)
 
-        if self.is_query(request.operation[TXN_TYPE]):
+        elif self.is_query(txn_type):
             self.process_query(request, frm)
-            return
+            self.total_read_request_number += 1
 
-        reply = self.getReplyFromLedger(ledger, request)
-        if reply:
-            logger.debug("{} returning REPLY from already processed "
-                         "REQUEST: {}".format(self, request))
-            self.transmitToClient(reply, frm)
-            return
+        elif self.is_txn_writable(txn_type):
+            reply = self.getReplyFromLedgerForRequest(request)
+            if reply:
+                logger.debug("{} returning REPLY from already processed "
+                             "REQUEST: {}".format(self, request))
+                self.transmitToClient(reply, frm)
+                return
 
-        # If the node is not already processing the request
-        if not self.isProcessingReq(request.key):
-            self.startedProcessingReq(request.key, frm)
-        # If not already got the propagate request(PROPAGATE) for the
-        # corresponding client request(REQUEST)
-        self.recordAndPropagate(request, frm)
-        self.send_ack_to_client((request.identifier, request.reqId), frm)
+            # If the node is not already processing the request
+            if not self.isProcessingReq(request.key):
+                self.startedProcessingReq(request.key, frm)
+                if request.isForced():
+                    # forced request should be processed before consensus
+                    req_handler = self.get_req_handler(txn_type=txn_type)
+                    req_handler.validate(request)
+                    req_handler.applyForced(request)
+
+            # If not already got the propagate request(PROPAGATE) for the
+            # corresponding client request(REQUEST)
+            self.recordAndPropagate(request, frm)
+            self.send_ack_to_client((request.identifier, request.reqId), frm)
+
+        else:
+            raise InvalidClientRequest(
+                request.identifier,
+                request.reqId,
+                'Pool is in readonly mode, try again in 60 seconds')
 
     def is_query(self, txn_type) -> bool:
         # Does the transaction type correspond to a read?
@@ -2167,6 +2181,9 @@ def is_query(self, txn_type) -> bool:
     def is_action(self, txn_type) -> bool:
         return txn_type in self.actionReqHandler.operation_types
 
+    def is_txn_writable(self, txn_type):
+        return True
+
     def process_query(self, request: Request, frm: str):
         # Process a read request from client
         handler = self.get_req_handler(txn_type=request.operation[TXN_TYPE])
@@ -2213,13 +2230,19 @@ def processPropagate(self, msg: Propagate, frm):
 
         if not self.isProcessingReq(request.key):
             ledger_id, seq_no = self.seqNoDB.get(request.key)
-            if seq_no is not None:
+            if ledger_id is not None and seq_no is not None:
                 logger.debug("{} ignoring propagated request {} "
                              "since it has been already ordered"
                              .format(self.name, msg))
                 return
 
             self.startedProcessingReq(request.key, clientName)
+            if request.isForced():
+                # forced request should be processed before consensus
+                req_handler = self.get_req_handler(
+                    txn_type=request.operation[TXN_TYPE])
+                req_handler.validate(request)
+                req_handler.applyForced(request)
 
         else:
             if clientName is not None and \
@@ -2264,17 +2287,20 @@ def handle_get_txn_req(self, request: Request, frm: str):
                                      'Invalid ledger id {}'.format(ledger_id),
                                      frm)
             return
+
         seq_no = request.operation.get(DATA)
         self.send_ack_to_client((request.identifier, request.reqId), frm)
         ledger = self.getLedger(ledger_id)
+
         try:
-            txn = self.getReplyFromLedger(ledger=ledger,
-                                          seq_no=seq_no)
+            txn = self.getReplyFromLedger(ledger, seq_no)
         except KeyError:
+            txn = None
+
+        if txn is None:
             logger.debug(
                 "{} can not handle GET_TXN request: ledger doesn't "
                 "have txn with seqNo={}". format(self, str(seq_no)))
-            txn = None
 
         result = {
             f.IDENTIFIER.nm: request.identifier,
@@ -3029,17 +3055,24 @@ def sendToNodes(self, msg: Any, names: Iterable[str] = None, message_splitter=No
         ) if r.name in names] if names else []
         self.send(msg, *rids, message_splitter=message_splitter)
 
-    def getReplyFromLedger(self, ledger, request=None, seq_no=None):
+    def getReplyFromLedgerForRequest(self, request):
+        ledger_id, seq_no = self.seqNoDB.get(request.digest)
+        if ledger_id is not None and seq_no is not None:
+            ledger = self.getLedger(ledger_id)
+            return self.getReplyFromLedger(ledger, seq_no)
+        else:
+            return None
+
+    def getReplyFromLedger(self, ledger, seq_no):
         # DoS attack vector, client requesting already processed request id
         # results in iterating over ledger (or its subset)
-        if seq_no is None:
-            ledger_id, seq_no = self.seqNoDB.get(request.digest)
-        if seq_no:
-            txn = ledger.getBySeqNo(int(seq_no))
-            if txn:
-                txn.update(ledger.merkleInfo(seq_no))
-                txn = self.update_txn_with_extra_data(txn)
-                return Reply(txn)
+        txn = ledger.getBySeqNo(int(seq_no))
+        if txn:
+            txn.update(ledger.merkleInfo(seq_no))
+            txn = self.update_txn_with_extra_data(txn)
+            return Reply(txn)
+        else:
+            return None
 
     def update_txn_with_extra_data(self, txn):
         """

plenum/test/node_request/test_already_processed_request.py

@@ -19,9 +19,9 @@ def get_method_call_count(method):
         assert len(counts) == 1
         return counts.pop()
 
-    def get_getReplyFromLedger_call_count():
+    def get_getReplyFromLedgerForRequest_call_count():
         return get_method_call_count(
-            next(iter(txnPoolNodeSet)).getReplyFromLedger)
+            next(iter(txnPoolNodeSet)).getReplyFromLedgerForRequest)
 
     def get_recordAndPropagate_call_count():
         return get_method_call_count(
@@ -30,13 +30,13 @@ def get_recordAndPropagate_call_count():
     def get_last_returned_val():
         rvs = []
         for node in txnPoolNodeSet:
-            rv = getAllReturnVals(node, node.getReplyFromLedger)
+            rv = getAllReturnVals(node, node.getReplyFromLedgerForRequest)
             rvs.append(rv[0])
         # All items are same in the list
         assert rvs.count(rvs[0]) == len(txnPoolNodeSet)
         return rvs[0]
 
-    rlc1 = get_getReplyFromLedger_call_count()
+    rlc1 = get_getReplyFromLedgerForRequest_call_count()
     rpc1 = get_recordAndPropagate_call_count()
 
     # Request which will be send twice
@@ -50,24 +50,24 @@ def get_last_returned_val():
         sdk_check_reply(req_res)
     first_req_id = request1[0][0]['reqId']
 
-    rlc2 = get_getReplyFromLedger_call_count()
+    rlc2 = get_getReplyFromLedgerForRequest_call_count()
     rpc2 = get_recordAndPropagate_call_count()
-    assert rlc2 - rlc1 == 1  # getReplyFromLedger was called
+    assert rlc2 - rlc1 == 1  # getReplyFromLedgerForRequest was called
     assert rpc2 - rpc1 == 1  # recordAndPropagate was called
     r1 = get_last_returned_val()
-    assert r1 is None  # getReplyFromLedger returned None since had not seen request
+    assert r1 is None  # getReplyFromLedgerForRequest returned None since had not seen request
 
     # Request which we will send only once
     request2 = sdk_send_random_and_check(looper, txnPoolNodeSet, sdk_pool_handle, sdk_wallet_client, 1)
     second_req_id = request2[0][0]['reqId']
 
     assert second_req_id != first_req_id
-    rlc3 = get_getReplyFromLedger_call_count()
+    rlc3 = get_getReplyFromLedgerForRequest_call_count()
     rpc3 = get_recordAndPropagate_call_count()
-    assert rlc3 - rlc2 == 1  # getReplyFromLedger was called again
+    assert rlc3 - rlc2 == 1  # getReplyFromLedgerForRequest was called again
     assert rpc3 - rpc2 == 1  # recordAndPropagate was called again
     r2 = get_last_returned_val()
-    assert r2 is None  # getReplyFromLedger returned None since had not seen request
+    assert r2 is None  # getReplyFromLedgerForRequest returned None since had not seen request
 
     # Reply for the first request, which is going to be sent again
     rep1 = request1[0][1]['result']
@@ -79,12 +79,12 @@ def get_last_returned_val():
     third_req_id = request3[0][0]['reqId']
 
     assert third_req_id == first_req_id
-    rlc4 = get_getReplyFromLedger_call_count()
+    rlc4 = get_getReplyFromLedgerForRequest_call_count()
     rpc4 = get_recordAndPropagate_call_count()
-    assert rlc4 - rlc3 == 1  # getReplyFromLedger was called again
+    assert rlc4 - rlc3 == 1  # getReplyFromLedgerForRequest was called again
     assert rpc4 - rpc3 == 0  # recordAndPropagate was not called
     r3 = get_last_returned_val()
-    # getReplyFromLedger did not return None this time since had seen request
+    # getReplyFromLedgerForRequest did not return None this time since had seen request
     assert r3 is not None
     rep3 = request3[0][1]['result']
 

plenum/test/node_request/test_req_idr_to_txn.py

@@ -21,17 +21,17 @@ def req_ids_to_txn(tconf):
 
 def test_req_id_to_txn_add(req_ids_to_txn):
     digest = "random_req_digest"
-    ledge_id = "1"
+    ledger_id = 1
     seq_no = 123
-    req_ids_to_txn.add(digest, ledge_id, seq_no)
+    req_ids_to_txn.add(digest, ledger_id, seq_no)
     new_ledger_id, new_seq_no = req_ids_to_txn.get(digest)
-    assert new_ledger_id == ledge_id
+    assert new_ledger_id == ledger_id
     assert seq_no == seq_no
 
 
 def test_req_id_to_txn_add_batch(req_ids_to_txn):
     batch = [("random_req_digest" + str(index),
-              "1",
+              1,
               123 + index)
              for index in range(3)]
     req_ids_to_txn.addBatch(batch)

plenum/test/test_node.py

@@ -335,6 +335,7 @@ def processRequest(self, request, frm):
                  Node.lost_master_primary,
                  Node.propose_view_change,
                  Node.getReplyFromLedger,
+                 Node.getReplyFromLedgerForRequest,
                  Node.recordAndPropagate,
                  Node.allLedgersCaughtUp,
                  Node.start_catchup,