refactor(side-dag): general improvements

Author: glevco

hathor/consensus/poa/poa_block_producer.py

@@ -17,6 +17,7 @@
 from typing import TYPE_CHECKING
 
 from structlog import get_logger
+from twisted.internet.base import DelayedCall
 from twisted.internet.interfaces import IDelayedCall
 from twisted.internet.task import LoopingCall
 
@@ -35,11 +36,8 @@
 
 logger = get_logger()
 
-# Number of seconds to wait for a sync to finish before trying to produce blocks
-_WAIT_SYNC_DELAY: int = 30
-
 # Number of seconds used between each signer depending on its distance to the expected signer
-_SIGNER_TURN_INTERVAL: int = 1
+_SIGNER_TURN_INTERVAL: int = 10
 
 
 class PoaBlockProducer:
@@ -54,8 +52,6 @@ class PoaBlockProducer:
         '_reactor',
         '_manager',
         '_poa_signer',
-        '_started_producing',
-        '_start_producing_lc',
         '_schedule_block_lc',
         '_last_seen_best_block',
         '_delayed_call',
@@ -71,10 +67,6 @@ def __init__(self, *, settings: HathorSettings, reactor: ReactorProtocol, poa_si
         self._poa_signer = poa_signer
         self._last_seen_best_block: Block | None = None
 
-        self._started_producing = False
-        self._start_producing_lc = LoopingCall(self._start_producing)
-        self._start_producing_lc.clock = self._reactor
-
         self._schedule_block_lc = LoopingCall(self._schedule_block)
         self._schedule_block_lc.clock = self._reactor
         self._delayed_call: IDelayedCall | None = None
@@ -89,13 +81,9 @@ def manager(self, manager: HathorManager) -> None:
         self._manager = manager
 
     def start(self) -> None:
-        self._start_producing_lc.start(_WAIT_SYNC_DELAY)
         self._schedule_block_lc.start(self._settings.AVG_TIME_BETWEEN_BLOCKS)
 
     def stop(self) -> None:
-        if self._start_producing_lc.running:
-            self._start_producing_lc.stop()
-
         if self._schedule_block_lc.running:
             self._schedule_block_lc.stop()
 
@@ -113,21 +101,15 @@ def _get_signer_index(self, previous_block: Block) -> int | None:
         except ValueError:
             return None
 
-    def _start_producing(self) -> None:
-        """Start producing new blocks."""
+    def _schedule_block(self) -> None:
+        """Schedule propagation of a new block."""
         if not self.manager.can_start_mining():
             # We're syncing, so we'll try again later
-            self._log.warn('cannot start producing new blocks, node not synced')
+            self._log.info('cannot produce new block, node not synced')
             return
 
-        self._log.info('started producing new blocks')
-        self._started_producing = True
-        self._start_producing_lc.stop()
-
-    def _schedule_block(self) -> None:
-        """Schedule propagation of a new block."""
         previous_block = self.manager.tx_storage.get_best_block()
-        if not self._started_producing or previous_block == self._last_seen_best_block:
+        if previous_block == self._last_seen_best_block:
             return
 
         self._last_seen_best_block = previous_block
@@ -139,6 +121,15 @@ def _schedule_block(self) -> None:
         expected_timestamp = self._expected_block_timestamp(previous_block, signer_index)
         propagation_delay = 0 if expected_timestamp < now else expected_timestamp - now
 
+        if self._delayed_call and self._delayed_call.active():
+            from hathor.transaction.poa import PoaBlock
+            assert isinstance(self._delayed_call, DelayedCall)
+            delayed_block = self._delayed_call.args[0]
+            assert isinstance(delayed_block, PoaBlock)
+            if delayed_block.weight != poa.BLOCK_WEIGHT_IN_TURN:
+                # we only cancel our delayed block if it was out of turn
+                self._delayed_call.cancel()
+
         self._delayed_call = self._reactor.callLater(propagation_delay, self._produce_block, previous_block)
         self._log.debug(
             'scheduling block production',
@@ -158,18 +149,18 @@ def _produce_block(self, previous_block: PoaBlock) -> None:
         self._poa_signer.sign_block(block)
         block.update_hash()
 
-        self.manager.on_new_tx(block, propagate_to_peers=False, fails_silently=False)
-        if not block.get_metadata().voided_by:
-            self.manager.connections.send_tx_to_peers(block)
-
-        self._log.debug(
+        self._log.info(
             'produced new block',
             block=block.hash_hex,
             height=block.get_height(),
             weight=block.weight,
             parent=block.get_block_parent_hash().hex(),
             voided=bool(block.get_metadata().voided_by),
         )
+        self.manager.on_new_tx(block, propagate_to_peers=False, fails_silently=False)
+        if not block.get_metadata().voided_by:
+            self.manager.connections.send_tx_to_peers(block)
+        self._delayed_call = None
 
     def _expected_block_timestamp(self, previous_block: Block, signer_index: int) -> int:
         """Calculate the expected timestamp for a new block."""
@@ -179,4 +170,7 @@ def _expected_block_timestamp(self, previous_block: Block, signer_index: int) ->
         index_distance = (signer_index - expected_index) % len(signers)
         assert 0 <= index_distance < len(signers)
         delay = _SIGNER_TURN_INTERVAL * index_distance
+        if index_distance > 0:
+            # if it's not our turn, we add a constant offset to the delay
+            delay += self._settings.AVG_TIME_BETWEEN_BLOCKS
         return previous_block.timestamp + self._settings.AVG_TIME_BETWEEN_BLOCKS + delay

tests/poa/test_poa_block_producer.py

@@ -57,7 +57,6 @@ def test_poa_block_producer_one_signer() -> None:
 
     # when we can start mining, we start producing blocks
     manager.can_start_mining = Mock(return_value=True)
-    reactor.advance(20)
 
     # we produce our first block
     reactor.advance(10)
@@ -116,7 +115,6 @@ def test_poa_block_producer_two_signers() -> None:
 
     # when we can start mining, we start producing blocks
     manager.can_start_mining = Mock(return_value=True)
-    reactor.advance(20)
 
     # we produce our first block
     reactor.advance(10)
@@ -144,14 +142,14 @@ def test_poa_block_producer_two_signers() -> None:
     manager.on_new_tx.reset_mock()
 
     # haven't produced the third block yet
-    reactor.advance(9)
+    reactor.advance(29)
 
     # we produce our third block
-    reactor.advance(2)
+    reactor.advance(1)
     manager.on_new_tx.assert_called_once()
     block3 = manager.on_new_tx.call_args.args[0]
     assert isinstance(block3, PoaBlock)
-    assert block3.timestamp == block2.timestamp + 11
+    assert block3.timestamp == block2.timestamp + 30
     assert block3.weight == poa.BLOCK_WEIGHT_OUT_OF_TURN
     assert block3.outputs == []
     assert block3.get_block_parent_hash() == block2.hash
@@ -161,15 +159,15 @@ def test_poa_block_producer_two_signers() -> None:
 @pytest.mark.parametrize(
     ['previous_height', 'signer_index', 'expected_delay'],
     [
-        (0, 0, 33),
+        (0, 0, 90),
         (0, 1, 30),
-        (0, 2, 31),
-        (0, 3, 32),
+        (0, 2, 70),
+        (0, 3, 80),
 
-        (1, 0, 32),
-        (1, 1, 33),
+        (1, 0, 80),
+        (1, 1, 90),
         (1, 2, 30),
-        (1, 3, 31),
+        (1, 3, 70),
     ]
 )
 def test_expected_block_timestamp(previous_height: int, signer_index: int, expected_delay: int) -> None:

tests/poa/test_poa_simulation.py

@@ -117,7 +117,7 @@ def test_one_producer_allowed(self) -> None:
 
         # manager is allowed to produce blocks, so it does
         manager.allow_mining_without_peers()
-        self.simulator.run(120)
+        self.simulator.run(90)
         assert manager.tx_storage.get_block_count() == 10
 
         _assert_height_weight_signer_id(
@@ -155,9 +155,9 @@ def test_two_producers(self) -> None:
         self.simulator.add_connection(connection)
 
         # both managers are producing blocks
-        self.simulator.run(125)
-        assert manager1.tx_storage.get_block_count() == 16
-        assert manager2.tx_storage.get_block_count() == 17
+        self.simulator.run(100)
+        assert manager1.tx_storage.get_block_count() == 15
+        assert manager2.tx_storage.get_block_count() == 16
         assert manager1.tx_storage.get_best_block_tips() == manager2.tx_storage.get_best_block_tips()
 
         _assert_height_weight_signer_id(
@@ -195,13 +195,10 @@ def test_two_producers(self) -> None:
 
             if height % 2 == 0:
                 # if the height is even, it's manager1's turn.
-                # manager2 will produce its block too, but it'll be voided and not propagated.
                 assert len(blocks_manager1) == 1
-                assert len(blocks_manager2) == 2
                 _assert_block_in_turn(blocks_manager1[0], signer1)
             else:
                 # if the height is odd, the opposite happens
-                assert len(blocks_manager1) == 2
                 assert len(blocks_manager2) == 1
                 _assert_block_in_turn(blocks_manager2[0], signer2)
 
@@ -214,41 +211,45 @@ def test_producer_leave_and_comeback(self) -> None:
         # out of turn
         manager1 = self._get_manager(signer1)
         manager1.allow_mining_without_peers()
-        self.simulator.run(60)
+        self.simulator.run(50)
 
         manager2 = self._get_manager(signer2)
         connection = FakeConnection(manager1, manager2)
         self.simulator.add_connection(connection)
         self.simulator.run(80)
 
         manager2.stop()
-        self.simulator.run(40)
+        self.simulator.run(70)
 
         manager2.start()
         self.simulator.run(30)
 
-        assert manager1.tx_storage.get_block_count() == 23
-        assert manager2.tx_storage.get_block_count() == 23
+        assert manager1.tx_storage.get_block_count() == 22
+        assert manager2.tx_storage.get_block_count() == 22
         assert manager1.tx_storage.get_best_block_tips() == manager2.tx_storage.get_best_block_tips()
 
         _assert_height_weight_signer_id(
             manager1.tx_storage.get_all_transactions(),
             [
+                # Before manager2 joins, only manager1 produces blocks
                 (1, poa.BLOCK_WEIGHT_OUT_OF_TURN, signer_id1),
                 (2, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
                 (3, poa.BLOCK_WEIGHT_OUT_OF_TURN, signer_id1),
                 (4, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
                 (5, poa.BLOCK_WEIGHT_OUT_OF_TURN, signer_id1),
                 (6, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
+                # When manager2 joins, both of them start taking turns
                 (7, poa.BLOCK_WEIGHT_IN_TURN, signer_id2),
                 (8, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
                 (9, poa.BLOCK_WEIGHT_IN_TURN, signer_id2),
                 (10, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
                 (11, poa.BLOCK_WEIGHT_IN_TURN, signer_id2),
                 (12, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
+                # manager2 leaves so manager1 produces all the next blocks
                 (13, poa.BLOCK_WEIGHT_OUT_OF_TURN, signer_id1),
                 (14, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
                 (15, poa.BLOCK_WEIGHT_OUT_OF_TURN, signer_id1),
+                # manager2 comes back again, so both of them take turns again
                 (16, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
                 (17, poa.BLOCK_WEIGHT_IN_TURN, signer_id2),
                 (18, poa.BLOCK_WEIGHT_IN_TURN, signer_id1),
@@ -272,7 +273,7 @@ def test_existing_storage(self) -> None:
         manager1 = artifacts1.manager
         manager1.allow_mining_without_peers()
 
-        self.simulator.run(80)
+        self.simulator.run(50)
         assert manager1.tx_storage.get_block_count() == 6
 
         _assert_height_weight_signer_id(
@@ -296,7 +297,7 @@ def test_existing_storage(self) -> None:
         manager2 = artifacts.manager
         manager2.allow_mining_without_peers()
 
-        self.simulator.run(80)
+        self.simulator.run(60)
         assert manager2.tx_storage.get_block_count() == 12
 
         _assert_height_weight_signer_id(
@@ -332,7 +333,7 @@ def test_new_signer_added(self) -> None:
         manager_1a = artifacts_1a.manager
         manager_1a.allow_mining_without_peers()
 
-        self.simulator.run(80)
+        self.simulator.run(50)
         assert manager_1a.tx_storage.get_block_count() == 6
 
         _assert_height_weight_signer_id(
@@ -358,7 +359,7 @@ def test_new_signer_added(self) -> None:
         manager_1b = artifacts_1b.manager
         manager_1b.allow_mining_without_peers()
 
-        self.simulator.run(80)
+        self.simulator.run(90)
         assert manager_1b.tx_storage.get_block_count() == 11
 
         # after we restart it, new blocks are alternating
@@ -383,7 +384,7 @@ def test_new_signer_added(self) -> None:
 
         connection = FakeConnection(manager_1b, manager_2)
         self.simulator.add_connection(connection)
-        self.simulator.run(40)
+        self.simulator.run(60)
 
         # it should sync to the same blockchain
         _assert_height_weight_signer_id(
@@ -463,7 +464,7 @@ def test_use_case(self) -> None:
 
         manager = self._get_manager(signer)
         manager.allow_mining_without_peers()
-        self.simulator.run(130)
+        self.simulator.run(100)
         assert manager.tx_storage.get_block_count() == 11
 
         _assert_height_weight_signer_id(