refactor(sync-v2): introduce all structures needed without any sync-v2

Author: jansegre

.github/workflows/main.yml

@@ -13,7 +13,7 @@ jobs:
   test:
     name: python-${{ matrix.python }} (${{ matrix.os }})
     runs-on: ${{ matrix.os }}
-    timeout-minutes: 40  # default is 360
+    timeout-minutes: 60  # default is 360
     strategy:
       fail-fast: false
       matrix:

Makefile

@@ -36,6 +36,10 @@ tests-doctests:
 tests-lib:
 	pytest --durations=10 $(pytest_flags) --doctest-modules hathor $(tests_lib)
 
+.PHONY: tests-quick
+tests-quick:
+	pytest --durations=10 $(pytest_flags) --doctest-modules hathor $(tests_lib) --maxfail=1
+
 .PHONY: tests-genesis
 tests-genesis:
 	HATHOR_TEST_CONFIG_FILE=hathor.conf.mainnet pytest tests/tx/test_genesis.py

hathor/cli/run_node.py

@@ -80,6 +80,8 @@ def create_parser(self) -> ArgumentParser:
         parser.add_argument('--max-output-script-size', type=int, default=None, help='Custom max accepted script size '
                             'on /push-tx API')
         parser.add_argument('--sentry-dsn', help='Sentry DSN')
+        parser.add_argument('--x-sync-bridge', action='store_true',
+                            help='Enable support for running both sync protocols. DO NOT ENABLE, IT WILL BREAK.')
         return parser
 
     def prepare(self, args: Namespace) -> None:
@@ -227,7 +229,8 @@ def create_wallet():
         network = settings.NETWORK_NAME
         self.manager = HathorManager(reactor, peer_id=peer_id, network=network, hostname=hostname,
                                      tx_storage=self.tx_storage, wallet=self.wallet, wallet_index=args.wallet_index,
-                                     stratum_port=args.stratum, ssl=True, checkpoints=settings.CHECKPOINTS)
+                                     stratum_port=args.stratum, ssl=True, checkpoints=settings.CHECKPOINTS,
+                                     enable_sync_v1=True, enable_sync_v2=args.x_sync_bridge)
         if args.allow_mining_without_peers:
             self.manager.allow_mining_without_peers()
 

hathor/consensus.py

@@ -154,14 +154,14 @@ def update_voided_info(self, block: Block) -> None:
 
         # Update accumulated weight of the transactions voiding us.
         assert block.hash not in voided_by
-        for h in voided_by:
+        for h in sorted(voided_by):
             tx = storage.get_transaction(h)
             tx_meta = tx.get_metadata()
             tx_meta.accumulated_weight = sum_weights(tx_meta.accumulated_weight, block.weight)
             storage.save_transaction(tx, only_metadata=True)
 
         # Check conflicts of the transactions voiding us.
-        for h in voided_by:
+        for h in sorted(voided_by):
             tx = storage.get_transaction(h)
             if not tx.is_block:
                 assert isinstance(tx, Transaction)

hathor/indexes.py

@@ -44,27 +44,29 @@ class IndexesManager:
         so it will know which index is better to use in each moment
     """
 
-    def __init__(self) -> None:
-        self.tips_index = TipsIndex()
+    def __init__(self, with_tips_index: bool = True) -> None:
+        self.tips_index = TipsIndex() if with_tips_index else None
         self.txs_index = TransactionsIndex()
 
     def add_tx(self, tx: BaseTransaction) -> bool:
         """ Add a transaction to the indexes
 
         :param tx: Transaction to be added
         """
-        r1 = self.tips_index.add_tx(tx)
-        r2 = self.txs_index.add_tx(tx)
-        assert r1 == r2
+        r1 = self.txs_index.add_tx(tx)
+        if self.tips_index is not None:
+            r2 = self.tips_index.add_tx(tx)
+            assert r1 == r2
         return r1
 
     def del_tx(self, tx: BaseTransaction, *, relax_assert: bool = False) -> None:
         """ Delete a transaction from the indexes
 
         :param tx: Transaction to be deleted
         """
-        self.tips_index.del_tx(tx, relax_assert=relax_assert)
         self.txs_index.del_tx(tx)
+        if self.tips_index is not None:
+            self.tips_index.del_tx(tx, relax_assert=relax_assert)
 
     def get_newest(self, count: int) -> Tuple[List[bytes], bool]:
         """ Get transactions or blocks in txs_index from the newest to the oldest

hathor/manager.py

@@ -41,7 +41,7 @@
 from hathor.transaction.exceptions import TxValidationError
 from hathor.transaction.storage import TransactionStorage
 from hathor.transaction.storage.exceptions import TransactionDoesNotExist
-from hathor.util import LogDuration, random_choices
+from hathor.util import LogDuration, not_none, random_choices
 from hathor.wallet import BaseWallet
 
 settings = HathorSettings()
@@ -67,6 +67,7 @@ def __init__(self, reactor: IReactorCore, peer_id: Optional[PeerId] = None, netw
                  wallet: Optional[BaseWallet] = None, tx_storage: Optional[TransactionStorage] = None,
                  peer_storage: Optional[Any] = None, default_port: int = 40403, wallet_index: bool = False,
                  stratum_port: Optional[int] = None, ssl: bool = True,
+                 enable_sync_v1: bool = True, enable_sync_v2: bool = False,
                  capabilities: Optional[List[str]] = None, checkpoints: Optional[List[Checkpoint]] = None,
                  rng: Optional[Rng] = None) -> None:
         """
@@ -81,7 +82,7 @@ def __init__(self, reactor: IReactorCore, peer_id: Optional[PeerId] = None, netw
         :param pubsub: If not given, a new one is created.
         :type pubsub: :py:class:`hathor.pubsub.PubSubManager`
 
-        :param tx_storage: If not given, a :py:class:`TransactionMemoryStorage` one is created.
+        :param tx_storage: Required storage backend.
         :type tx_storage: :py:class:`hathor.transaction.storage.transaction_storage.TransactionStorage`
 
         :param peer_storage: If not given, a new one is created.
@@ -99,7 +100,12 @@ def __init__(self, reactor: IReactorCore, peer_id: Optional[PeerId] = None, netw
         from hathor.metrics import Metrics
         from hathor.p2p.factory import HathorClientFactory, HathorServerFactory
         from hathor.p2p.manager import ConnectionsManager
-        from hathor.transaction.storage.memory_storage import TransactionMemoryStorage
+
+        if not (enable_sync_v1 or enable_sync_v2):
+            raise TypeError(f'{type(self).__name__}() at least one sync version is required')
+
+        if tx_storage is None:
+            raise TypeError(f'{type(self).__name__}() missing 1 required positional argument: \'tx_storage\'')
 
         self.log = logger.new()
 
@@ -138,7 +144,7 @@ def __init__(self, reactor: IReactorCore, peer_id: Optional[PeerId] = None, netw
 
         # XXX Should we use a singleton or a new PeerStorage? [msbrogli 2018-08-29]
         self.pubsub = pubsub or PubSubManager(self.reactor)
-        self.tx_storage = tx_storage or TransactionMemoryStorage()
+        self.tx_storage = tx_storage
         self.tx_storage.pubsub = self.pubsub
         if wallet_index and self.tx_storage.with_index:
             self.tx_storage.wallet_index = WalletIndex(self.pubsub)
@@ -156,10 +162,12 @@ def __init__(self, reactor: IReactorCore, peer_id: Optional[PeerId] = None, netw
         self.peer_discoveries: List[PeerDiscovery] = []
 
         self.ssl = ssl
-        self.server_factory = HathorServerFactory(self.network, self.my_peer, node=self, use_ssl=ssl)
-        self.client_factory = HathorClientFactory(self.network, self.my_peer, node=self, use_ssl=ssl)
+        self.server_factory = HathorServerFactory(self.network, self.my_peer, node=self, use_ssl=ssl,
+                                                  enable_sync_v1=enable_sync_v1, enable_sync_v2=enable_sync_v2)
+        self.client_factory = HathorClientFactory(self.network, self.my_peer, node=self, use_ssl=ssl,
+                                                  enable_sync_v1=enable_sync_v1, enable_sync_v2=enable_sync_v2)
         self.connections = ConnectionsManager(self.reactor, self.my_peer, self.server_factory, self.client_factory,
-                                              self.pubsub, self, ssl, rng=self.rng)
+                                              self.pubsub, self, ssl, whitelist_only=False, rng=self.rng)
 
         self.wallet = wallet
         if self.wallet:
@@ -193,6 +201,8 @@ def __init__(self, reactor: IReactorCore, peer_id: Optional[PeerId] = None, netw
         # List of capabilities of the peer
         if capabilities is not None:
             self.capabilities = capabilities
+        elif enable_sync_v2:
+            self.capabilities = [settings.CAPABILITY_WHITELIST, settings.CAPABILITY_SYNC_V2]
         else:
             self.capabilities = [settings.CAPABILITY_WHITELIST]
 
@@ -507,19 +517,33 @@ def generate_parent_txs(self, timestamp: Optional[float]) -> 'ParentTxs':
         This method tries to return a stable result, such that for a given timestamp and storage state it will always
         return the same.
         """
-        if timestamp is None:
-            timestamp = self.reactor.seconds()
-        can_include_intervals = sorted(self.tx_storage.get_tx_tips(timestamp - 1))
-        assert can_include_intervals, 'tips cannot be empty'
-        max_timestamp = max(int(i.begin) for i in can_include_intervals)
-        must_include: List[bytes] = []
-        assert len(can_include_intervals) > 0, f'invalid timestamp "{timestamp}", no tips found"'
-        if len(can_include_intervals) < 2:
-            # If there is only one tip, let's randomly choose one of its parents.
-            must_include_interval = can_include_intervals[0]
-            must_include = [must_include_interval.data]
-            can_include_intervals = sorted(self.tx_storage.get_tx_tips(must_include_interval.begin - 1))
-        can_include = [i.data for i in can_include_intervals]
+        # get all that are before "timestamp"
+        # XXX: maybe add some tolerance?
+        tx_tips = list(self.tx_storage.iter_tx_tips(timestamp))
+        must_include: List[bytes]
+        can_include: List[bytes]
+        max_timestamp: int
+        if not tx_tips:
+            # there are no txs on mempool, repeat the same tx parents as the best block
+            must_include = []
+            # can_include = self.tx_storage.get_best_block().parents[1:]
+            best_block = self.tx_storage.get_best_block()
+            if best_block.is_genesis:
+                can_include = [settings.GENESIS_TX1_HASH, settings.GENESIS_TX2_HASH]
+            else:
+                can_include = self.tx_storage.get_best_block().parents[1:]
+            max_timestamp = max(tx.timestamp for tx in map(self.tx_storage.get_transaction, can_include) if tx)
+        elif len(tx_tips) < 2:
+            # there is only one tx, it must be included, and either of its parents can be included
+            only_tx = tx_tips[0]
+            must_include = [not_none(tx.hash) for tx in tx_tips]
+            can_include = only_tx.parents[:]
+            max_timestamp = only_tx.timestamp
+        else:
+            # otherwise we can include any tx on the mempool
+            must_include = []
+            can_include = [not_none(tx.hash) for tx in tx_tips]
+            max_timestamp = max(tx.timestamp for tx in tx_tips)
         return ParentTxs(max_timestamp, can_include, must_include)
 
     def allow_mining_without_peers(self) -> None:
@@ -646,6 +670,17 @@ def _make_block_template(self, parent_block: Block, parent_txs: 'ParentTxs', cur
             score=sum_weights(parent_block_metadata.score, weight),
         )
 
+    def get_current_score(self, parent_block: Optional[Block] = None, timestamp: Optional[int] = None) -> float:
+        if timestamp is None:
+            timestamp = int(max(self.tx_storage.latest_timestamp, self.reactor.seconds()))
+        if parent_block is None:
+            block = self.tx_storage.get_transaction(next(iter(self.tx_storage.get_best_block_tips())))
+            assert isinstance(block, Block)
+            parent_block = block
+        parent_block_metadata = parent_block.get_metadata()
+        weight = daa.calculate_next_weight(parent_block, timestamp)
+        return sum_weights(parent_block_metadata.score, weight)
+
     def generate_mining_block(self, timestamp: Optional[int] = None,
                               parent_block_hash: Optional[bytes] = None,
                               data: bytes = b'', address: Optional[bytes] = None,

hathor/p2p/downloader.py

@@ -26,9 +26,9 @@
 settings = HathorSettings()
 
 if TYPE_CHECKING:
-    from hathor.manager import HathorManager  # noqa: F401
-    from hathor.p2p.node_sync import NodeSyncTimestamp  # noqa: F401
-    from hathor.transaction import BaseTransaction  # noqa: F401
+    from hathor.manager import HathorManager
+    from hathor.p2p.node_sync import NodeSyncTimestamp
+    from hathor.transaction import BaseTransaction
 
 logger = get_logger()
 

hathor/p2p/factory.py

@@ -42,13 +42,20 @@ def __init__(
             *,
             node: 'HathorManager',
             use_ssl: bool,
+            enable_sync_v1: bool,
+            enable_sync_v2: bool,
     ):
+        if not (enable_sync_v1 or enable_sync_v2):
+            raise ValueError('At least one sync version is required')
+
         super().__init__()
         self.network = network
         self.my_peer = my_peer
         self.connections = connections
         self.node = node
         self.use_ssl = use_ssl
+        self.enable_sync_v1 = enable_sync_v1
+        self.enable_sync_v2 = enable_sync_v2
 
     def buildProtocol(self, addr: Tuple[str, int]) -> MyServerProtocol:
         p = self.protocol(
@@ -58,6 +65,8 @@ def buildProtocol(self, addr: Tuple[str, int]) -> MyServerProtocol:
             node=self.node,
             use_ssl=self.use_ssl,
             inbound=True,
+            enable_sync_v1=self.enable_sync_v1,
+            enable_sync_v2=self.enable_sync_v2,
         )
         p.factory = self
         return p
@@ -78,13 +87,20 @@ def __init__(
             *,
             node: 'HathorManager',
             use_ssl: bool,
+            enable_sync_v1: bool,
+            enable_sync_v2: bool,
     ):
+        if not (enable_sync_v1 or enable_sync_v2):
+            raise ValueError('At least one sync version is required')
+
         super().__init__()
         self.network = network
         self.my_peer = my_peer
         self.connections = connections
         self.node = node
         self.use_ssl = use_ssl
+        self.enable_sync_v1 = enable_sync_v1
+        self.enable_sync_v2 = enable_sync_v2
 
     def buildProtocol(self, addr: Tuple[str, int]) -> MyClientProtocol:
         p = self.protocol(
@@ -94,6 +110,8 @@ def buildProtocol(self, addr: Tuple[str, int]) -> MyClientProtocol:
             node=self.node,
             use_ssl=self.use_ssl,
             inbound=False,
+            enable_sync_v1=self.enable_sync_v1,
+            enable_sync_v2=self.enable_sync_v2,
         )
         p.factory = self
         return p

hathor/p2p/manager.py

@@ -52,10 +52,11 @@ class ConnectionsManager:
     connected_peers: Dict[str, HathorProtocol]
     connecting_peers: Dict[IStreamClientEndpoint, Deferred]
     handshaking_peers: Set[HathorProtocol]
+    whitelist_only: bool
 
     def __init__(self, reactor: ReactorBase, my_peer: PeerId, server_factory: 'HathorServerFactory',
                  client_factory: 'HathorClientFactory', pubsub: PubSubManager, manager: 'HathorManager',
-                 ssl: bool, rng: Rng) -> None:
+                 ssl: bool, whitelist_only: bool, rng: Rng) -> None:
         self.log = logger.new()
         self.rng = rng
 
@@ -106,6 +107,9 @@ def __init__(self, reactor: ReactorBase, my_peer: PeerId, server_factory: 'Hatho
 
         self.ssl = ssl
 
+        # Parameter to explicitly enable whitelist-only mode, when False it will still check the whitelist for sync-v1
+        self.whitelist_only = whitelist_only
+
     def start(self) -> None:
         self.lc_reconnect.start(5, now=False)
         if settings.ENABLE_PEER_WHITELIST:

hathor/p2p/messages.py

@@ -90,10 +90,25 @@ class ProtocolMessages(Enum):
     GET_NEXT = 'GET-NEXT'
     NEXT = 'NEXT'
 
-    GET_BLOCKS = 'GET-BLOCKS'  # Request a list of hashes for blocks. Payload is the current latest block.
-    BLOCKS = 'BLOCKS'  # Send a list of hashes for blocks. Payload is a list of hashes.
+    # Sync-v2 messages
 
-    GET_TRANSACTIONS = 'GET-TRANSACTIONS'  # Request a list of hashes for transactions.
-    TRANSACTIONS = 'TRANSACTIONS'  # Send a list of hashes for transactions.
+    GET_NEXT_BLOCKS = 'GET-NEXT-BLOCKS'
+    BLOCKS = 'BLOCKS'
+    BLOCKS_END = 'BLOCKS-END'
 
-    HASHES = 'HASHES'
+    GET_BEST_BLOCK = 'GET-BEST-BLOCK'  # Request the best block of the peer
+    BEST_BLOCK = 'BEST-BLOCK'  # Send the best block to your peer
+
+    GET_BLOCK_TXS = 'GET-BLOCK-TXS'  # TODO: rename, maybe GET-TX-RANGE or repurpose GET-TRANSACTIONS above
+    TRANSACTION = 'TRANSACTION'
+
+    GET_MEMPOOL = 'GET-MEMPOOL'  # TODO: rename, maybe GET-TX-RANGE or repurpose GET-TRANSACTIONS above
+    MEMPOOL_END = 'MEMPOOL-END'  # End of mempool sync
+
+    GET_COMMON_CHAIN = 'GET-COMMON-CHAIN'
+    COMMON_CHAIN = 'COMMON-CHAIN'
+
+    GET_PEER_BLOCK_HASHES = 'GET-PEER-BLOCK-HASHES'
+    PEER_BLOCK_HASHES = 'PEER-BLOCK-HASHES'
+
+    STOP_BLOCK_STREAMING = 'STOP-BLOCK-STREAMING'