Merge branch 'master' of github.com:hyperledger/indy-plenum

Author: Nikita Khateev

docs/source/diagrams/catchup-procedure.puml

@@ -239,7 +239,7 @@ opt last_caught_up_3PC > Master's last_ordered_3pc
             note right Node4
                 <b>Clear following replica's collections:
                     - batches <i>- up to last_caught_up_3PC,
-                    - sentPrePrepares <i>- up to last_caught_up_3PC,
+                    - sent_preprepares <i>- up to last_caught_up_3PC,
                     - prePrepares <i>- up to last_caught_up_3PC,
                     - prepares <i>- up to last_caught_up_3PC,
                     - commits <i>- up to last_caught_up_3PC,
@@ -275,7 +275,7 @@ opt last_caught_up_3PC > Master's last_ordered_3pc
                 note right Node4
                     <b>Clear following replica's collections completely:
                         - batches,
-                        - sentPrePrepares,
+                        - sent_preprepares,
                         - prePrepares,
                         - prepares,
                         - commits,

docs/source/diagrams/checkpoints.puml

@@ -217,7 +217,7 @@ note right of Node4
     <b>Garbage collection</b> removes messages up to
     (current view_no, stabilized checkpoint seq_no_end) from following replica's
     collections:
-        - sentPrePrepares,
+        - sent_preprepares,
         - prePrepares,
         - prepares,
         - commits,

plenum/common/constants.py

@@ -22,6 +22,8 @@
 PROPAGATE = "PROPAGATE"
 
 PREPREPARE = "PREPREPARE"
+OLD_VIEW_PREPREPARE_REQ = "OLD_VIEW_PREPREPARE_REQ"
+OLD_VIEW_PREPREPARE_REP = "OLD_VIEW_PREPREPARE_REP"
 PREPARE = "PREPARE"
 COMMIT = "COMMIT"
 CHECKPOINT = "CHECKPOINT"

plenum/common/exceptions.py

@@ -4,6 +4,7 @@
 #  - review the list and remove obsolete ones
 #  - refactor hierarchy of exceptions taking into account ones
 #    from common/exceptions.py
+from typing import Any, Callable
 
 from common.exceptions import LogicError
 
@@ -204,6 +205,15 @@ class MismatchedMessageReplyException(InvalidNodeMsg):
     pass
 
 
+class IncorrectMessageForHandlingException(InvalidNodeMsg):
+    def __init__(self, msg: Any, reason: str, log_method: Callable, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.log_method = log_method
+        self.msg = msg
+        self.reason = reason
+        self.args = args
+
+
 class MissingNodeOp(InvalidNodeMsg):
     pass
 

plenum/common/messages/internal_messages.py

@@ -1,4 +1,4 @@
-from typing import NamedTuple, List, Any
+from typing import NamedTuple, List, Any, Optional
 
 from plenum.common.exceptions import SuspiciousNode
 
@@ -22,8 +22,7 @@
                                 ('caught_up_till_3pc', tuple)])
 
 CheckpointStabilized = NamedTuple('CheckpointStabilized',
-                                  [('inst_id', int),
-                                   ('last_stable_3pc', tuple)])
+                                  [('last_stable_3pc', tuple)])
 
 RaisedSuspicion = NamedTuple('RaisedSuspicion',
                              [('inst_id', int),
@@ -32,6 +31,13 @@
 PreSigVerification = NamedTuple('PreSigVerification',
                                 [('cmsg', Any)])
 
+Missing3pcMessage = NamedTuple('Missing3pcMessage',
+                               [('msg_type', str),
+                                ('three_pc_key', tuple),
+                                ('inst_id', int),
+                                ('dst', List[str]),
+                                ('stash_data', Optional[tuple])])
+
 # by default view_no for StartViewChange is None meaning that we move to the next view
 NeedViewChange = NamedTuple('StartViewChange',
                             [('view_no', int)])

plenum/common/messages/node_messages.py

@@ -8,7 +8,7 @@
     CATCHUP_REP, VIEW_CHANGE_DONE, CURRENT_STATE, \
     MESSAGE_REQUEST, MESSAGE_RESPONSE, OBSERVED_DATA, BATCH_COMMITTED, OPERATION_SCHEMA_IS_STRICT, \
     BACKUP_INSTANCE_FAULTY, VIEW_CHANGE_START, PROPOSED_VIEW_NO, VIEW_CHANGE_CONTINUE, VIEW_CHANGE, VIEW_CHANGE_ACK, \
-    NEW_VIEW
+    NEW_VIEW, OLD_VIEW_PREPREPARE_REQ, OLD_VIEW_PREPREPARE_REP
 from plenum.common.messages.client_request import ClientMessageValidator
 from plenum.common.messages.fields import NonNegativeNumberField, IterableField, \
     SerializedValueField, SignatureField, TieAmongField, AnyValueField, TimestampField, \
@@ -159,6 +159,23 @@ def _post_process(self, input_as_dict: Dict) -> Dict:
         return input_as_dict
 
 
+# TODO: use generic MessageReq mechanism once it's separated into an independent service
+class OldViewPrePrepareRequest(MessageBase):
+    typename = OLD_VIEW_PREPREPARE_REQ
+    schema = (
+        (f.INST_ID.nm, NonNegativeNumberField()),
+        (f.BATCH_IDS.nm, IterableField(BatchIDField())),
+    )
+
+
+class OldViewPrePrepareReply(MessageBase):
+    typename = OLD_VIEW_PREPREPARE_REP
+    schema = (
+        (f.INST_ID.nm, NonNegativeNumberField()),
+        (f.PREPREPARES.nm, IterableField(AnyField())),
+    )
+
+
 class Prepare(MessageBase):
     typename = PREPARE
     schema = (
@@ -233,8 +250,8 @@ class ViewChange(MessageBase):
     schema = (
         (f.VIEW_NO.nm, NonNegativeNumberField()),
         (f.STABLE_CHECKPOINT.nm, NonNegativeNumberField()),
-        (f.PREPARED.nm, IterableField(BatchIDField())),  # list of tuples (view_no, pp_seq_no, pp_digest)
-        (f.PREPREPARED.nm, IterableField(BatchIDField())),  # list of tuples (view_no, pp_seq_no, pp_digest)
+        (f.PREPARED.nm, IterableField(BatchIDField())),  # list of tuples (view_no, pp_view_no, pp_seq_no, pp_digest)
+        (f.PREPREPARED.nm, IterableField(BatchIDField())),  # list of tuples (view_no, pp_view_no, pp_seq_no, pp_digest)
         (f.CHECKPOINTS.nm, IterableField(AnyField()))  # list of Checkpoints TODO: should we change to tuples?
     )
 
@@ -254,7 +271,7 @@ class NewView(MessageBase):
         (f.VIEW_NO.nm, NonNegativeNumberField()),
         (f.VIEW_CHANGES.nm, IterableField(ViewChangeField())),  # list of tuples (node_name, view_change_digest)
         (f.CHECKPOINT.nm, AnyField()),  # Checkpoint to be selected as stable (TODO: or tuple?)
-        (f.BATCHES.nm, IterableField(BatchIDField()))  # list of tuples (view_no, pp_seq_no, pp_digest)
+        (f.BATCHES.nm, IterableField(BatchIDField()))  # list of tuples (view_no, pp_view_no, pp_seq_no, pp_digest)
         # that should get into new view
     )
 

plenum/common/types.py

@@ -99,11 +99,13 @@ class f:  # provides a namespace for reusable field constants
     TAA_ACCEPTANCE_TIME = Field("time", float)
     # View change
     STABLE_CHECKPOINT = Field("stableCheckpoint", int)
-    PREPARED = Field("prepared", List[Tuple[int, int, str]])         # view_no, pp_seq_no, pp_digest
-    PREPREPARED = Field("preprepared", List[Tuple[int, int, str]])   # view_no, pp_seq_no, pp_digest
-    CHECKPOINTS = Field("checkpoints", List[Any])                    # Any ==Checkpoint
+    PREPARED = Field("prepared", List[Tuple[int, int, int, str]])         # view_no, pp_view_no, pp_seq_no, pp_digest
+    PREPREPARED = Field("preprepared", List[Tuple[int, int, int, str]])   # view_no, pp_view_no, pp_seq_no, pp_digest
+    BATCH_IDS = Field("batch_ids", List[Tuple[int, int, int, str]])  # view_no, pp_view_no, pp_seq_no, pp_digest
+    PREPREPARES = Field("preprepares", Any)                          # Any == PrePrepare
+    CHECKPOINTS = Field("checkpoints", List[Any])                    # Any == Checkpoint
     VIEW_CHANGES = Field("viewChanges", List[Tuple[str, str]])       # name, vc_digest
-    CHECKPOINT = Field("checkpoint", Any)                            # Any ==Checkpoint
+    CHECKPOINT = Field("checkpoint", Any)                            # Any == Checkpoint
     BATCHES = Field("batches", List[Tuple[int, int, str]])           # view_no, pp_seq_no, pp_digest
 
 

plenum/server/batch_handlers/three_pc_batch.py

@@ -1,6 +1,7 @@
 from plenum.common.ledger import Ledger
 from plenum.common.request import Request
 from plenum.common.types import f
+from plenum.server.consensus.consensus_shared_data import get_original_viewno
 
 
 class ThreePcBatch:
@@ -41,7 +42,7 @@ def from_pre_prepare(pre_prepare, state_root, txn_root, primaries, valid_digests
                             primaries=primaries,
                             valid_digests=valid_digests,
                             has_audit_txn=f.AUDIT_TXN_ROOT_HASH.nm in pre_prepare and pre_prepare.auditTxnRootHash is not None,
-                            original_view_no=pre_prepare.originalViewNo if f.ORIGINAL_VIEW_NO.nm in pre_prepare else None)
+                            original_view_no=get_original_viewno(pre_prepare))
 
     @staticmethod
     def from_ordered(ordered):

plenum/server/consensus/checkpoint_service.py

@@ -207,7 +207,7 @@ def _mark_checkpoint_stable(self, pp_seq_no):
         self.set_watermarks(low_watermark=pp_seq_no)
 
         self._remove_received_checkpoints(till_3pc_key=(self.view_no, pp_seq_no))
-        self._bus.send(CheckpointStabilized(self._data.inst_id, (self.view_no, pp_seq_no)))  # call OrderingService.l_gc()
+        self._bus.send(CheckpointStabilized((self.view_no, pp_seq_no)))  # call OrderingService.gc()
         self._logger.info("{} marked stable checkpoint {}".format(self, pp_seq_no))
 
     def set_watermarks(self, low_watermark: int, high_watermark: int = None):

plenum/server/consensus/consensus_shared_data.py

@@ -5,6 +5,8 @@
 from sortedcontainers import SortedListWithKey
 
 from plenum.common.startable import Mode
+from plenum.common.util import SortedDict
+from plenum.server.models import Prepares, Commits
 from plenum.common.types import f
 from plenum.server.propagator import Requests
 from plenum.server.quorums import Quorums
@@ -20,8 +22,12 @@
 BatchID = NamedTuple('BatchID', [('view_no', int), ('pp_view_no', int), ('pp_seq_no', int), ('pp_digest', str)])
 
 
+def get_original_viewno(pp):
+    return pp.originalViewNo if f.ORIGINAL_VIEW_NO.nm in pp else pp.viewNo
+
+
 def preprepare_to_batch_id(pre_prepare: PrePrepare) -> BatchID:
-    pp_view_no = pre_prepare.originalViewNo if f.ORIGINAL_VIEW_NO.nm in pre_prepare else pre_prepare.viewNo
+    pp_view_no = get_original_viewno(pre_prepare)
     return BatchID(pre_prepare.viewNo, pp_view_no, pre_prepare.ppSeqNo, pre_prepare.digest)
 
 
@@ -74,6 +80,28 @@ def __init__(self, name: str, validators: List[str], inst_id: int, is_master: bo
         self.legacy_last_prepared_before_view_change = None
         self.prev_view_prepare_cert = None
 
+        # Dictionary of sent PRE-PREPARE that are stored by primary replica
+        # which it has broadcasted to all other non primary replicas
+        # Key of dictionary is a 2 element tuple with elements viewNo,
+        # pre-prepare seqNo and value is the received PRE-PREPARE
+        self.sent_preprepares = SortedDict(lambda k: (k[0], k[1]))
+        # type: Dict[Tuple[int, int], PrePrepare]
+
+        # Dictionary of all Prepare requests. Key of dictionary is a 2
+        # element tuple with elements viewNo, seqNo and value is a 2 element
+        # tuple containing request digest and set of sender node names(sender
+        # replica names in case of multiple protocol instances)
+        # (viewNo, seqNo) -> ((identifier, reqId), {senders})
+        self.prepares = Prepares()
+        # type: Dict[Tuple[int, int], Tuple[Tuple[str, int], Set[str]]]
+
+        self.commits = Commits()
+        # type: Dict[Tuple[int, int], Tuple[Tuple[str, int], Set[str]]]
+
+        # Tracks for which keys PRE-PREPAREs have been requested.
+        # Cleared in `gc`
+        self.requested_pre_prepares = {}
+
     @property
     def name(self) -> str:
         return self._name