process_block_confirmations_task drafted

Author: dmugtasimov

node/blockchain/facade.py

@@ -260,3 +260,11 @@ def get_primary_validator(self) -> Optional[Node]:
     def yield_nodes(roles=None):
         for node in ORMNode.objects.filter_by_roles(roles):
             yield node.get_node()
+
+    @staticmethod
+    def get_confirmation_validator_identifiers():
+        return ORMNode.objects.filter_by_roles((NodeRole.CONFIRMATION_VALIDATOR,)).values_list('_id', flat=True)
+
+    @staticmethod
+    def is_confirmation_validator(identifier):
+        return ORMNode.objects.is_confirmation_validator(identifier)

node/blockchain/inner_models/block_confirmation.py

@@ -6,7 +6,7 @@
 from node.core.utils.cryptography import derive_public_key
 from node.core.utils.types import non_negative_int
 
-from ..types import AccountNumber, Hash, NodeRole, Signature
+from ..types import AccountNumber, Hash, Signature
 from .base import BaseModel
 
 
@@ -42,8 +42,7 @@ def validate_signer(self, blockchain_facade):
         if not blockchain_facade.has_blocks():
             return
 
-        cv_identifiers = (node.identifier for node in blockchain_facade.yield_nodes({NodeRole.CONFIRMATION_VALIDATOR}))
-        if self.signer not in cv_identifiers:
+        if not blockchain_facade.is_confirmation_validator(self.signer):
             raise ValidationError('Invalid block signer')
 
     def validate_business_logic(self):

node/blockchain/models/node.py

@@ -58,6 +58,9 @@ class NodeManager(DjongoManager.from_queryset(NodeQuerySet)):  # type: ignore
     def get_queryset(self):
         return super().get_queryset().filter(node__isnull=False)
 
+    def is_confirmation_validator(self, identifier):
+        return self.filter_by_roles((NodeRole.CONFIRMATION_VALIDATOR,)).filter(_id=identifier).exists()
+
 
 class Node(AccountState):
 

node/blockchain/tasks/process_block_confirmations.py

@@ -1,16 +1,57 @@
+from itertools import groupby
+from operator import attrgetter
+
 from celery import shared_task
+from django.db import transaction
+
+from node.blockchain.constants import BLOCK_LOCK
+from node.blockchain.facade import BlockchainFacade
+from node.blockchain.models import BlockConfirmation, PendingBlock
+from node.blockchain.utils.lock import lock
+
+
+@lock(BLOCK_LOCK)
+def process_next_block():
+    facade = BlockchainFacade.get_instance()
+    next_block_number = facade.get_next_block_number()
+    cv_identifiers = facade.get_confirmation_validator_identifiers()
+    confirmations = BlockConfirmation.objects.filter(number=next_block_number, signer__in=cv_identifiers)
+    grouped_confirmations = groupby(confirmations.order_by('hash'), key=attrgetter('hash'))
+    minimum_consensus = len(cv_identifiers) * 2 / 3
+
+    finalizable_hashes = [
+        hash_ for hash_, _confirmations in grouped_confirmations if len(list(_confirmations)) >= minimum_consensus
+    ]
+
+    if not finalizable_hashes:
+        return False
+
+    if len(finalizable_hashes) >= 2:
+        # We should never get here
+        raise ValueError('More than one finalizable hash found')
+
+    assert len(finalizable_hashes) == 1
+    hash_ = finalizable_hashes[0]
+
+    pending_block = PendingBlock.objects.get_or_none(number=next_block_number, hash=hash_)
+    if pending_block is None:
+        # TODO(dmu) CRITICAL: https://thenewboston.atlassian.net/browse/BC-283
+        raise NotImplementedError('Edge case of processing confirmed missing pending block is not implemented')
+
+    with transaction.atomic():
+        facade.add_block_from_json(pending_block.body, expect_locked=True)
+        # There may be blocks with other hashes therefore we delete all of them
+        PendingBlock.objects.filter(number__lte=next_block_number).delete()
+        BlockConfirmation.objects.filter(number__lte=next_block_number).delete()
 
-# from node.blockchain.facade import BlockchainFacade
+    return True
 
 
 @shared_task
 def process_block_confirmations_task():
-    # TODO(dmu) CRITICAL: Implement https://thenewboston.atlassian.net/browse/BC-272
-    # facade = BlockchainFacade.get_instance()
-    # next_block_number = facade.get_next_block_number()
-    # get all confirmation for next block number
-    # if there is no 2/3
-    raise NotImplementedError
+    should_process_next_block = True
+    while should_process_next_block:
+        should_process_next_block = process_next_block()
 
 
 def start_process_block_confirmations_task():