Major refactor of Request/Response cycle

- Get a small, simple request loop definition
- Component methods have access to a lot less object state
- Unfortunately it split things even a bit more, into:
  - Exchanges
  - Normalizers
  - Validators
- Get proper type checking on the call

Author: carver

p2p/peer.py

@@ -478,7 +478,7 @@ def decrypt_header(self, data: bytes) -> bytes:
         self.ingress_mac.update(sxor(aes, header_ciphertext))
         expected_header_mac = self.ingress_mac.digest()[:HEADER_LEN]
         if not bytes_eq(expected_header_mac, header_mac):
-            raise DecryptionError('Invalid header mac: expected %s, got %s'.format(
+            raise DecryptionError('Invalid header mac: expected {}, got {}'.format(
                 expected_header_mac, header_mac))
         return self.aes_dec.update(header_ciphertext)
 
@@ -1066,8 +1066,8 @@ async def request_stuff() -> None:
         hashes = tuple(header.hash for header in headers)
         if peer_class == ETHPeer:
             peer = cast(ETHPeer, peer)
-            peer.sub_proto._send_get_block_bodies(hashes)
-            peer.sub_proto._send_get_receipts(hashes)
+            peer.sub_proto.send_get_block_bodies(hashes)
+            peer.sub_proto.send_get_receipts(hashes)
         else:
             peer = cast(LESPeer, peer)
             request_id = 1

p2p/protocol.py

@@ -1,11 +1,14 @@
+from abc import ABC, abstractmethod
 import logging
 import struct
 from typing import (
     Any,
     Dict,
+    Generic,
     List,
     Tuple,
     Type,
+    TypeVar,
     TYPE_CHECKING,
     Union,
 )
@@ -123,6 +126,27 @@ def encode(self, data: _DecodedMsgType) -> Tuple[bytes, bytes]:
         return header, body
 
 
+TCommandPayload = TypeVar('TCommandPayload', bound=_DecodedMsgType)
+
+
+class BaseRequest(ABC, Generic[TCommandPayload]):
+    """
+    Must define command_payload during init. This is the data that will
+    be sent to the peer with the request command.
+    """
+    command_payload: TCommandPayload
+
+    @property
+    @abstractmethod
+    def cmd_type(self) -> Type[Command]:
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def response_type(self) -> Type[Command]:
+        raise NotImplementedError
+
+
 class Protocol:
     logger = logging.getLogger("p2p.protocol.Protocol")
     name: str = None
@@ -135,11 +159,20 @@ def __init__(self, peer: 'BasePeer', cmd_id_offset: int) -> None:
         self.peer = peer
         self.cmd_id_offset = cmd_id_offset
         self.commands = [cmd_class(cmd_id_offset) for cmd_class in self._commands]
+        self.cmd_by_type = {cmd_class: cmd_class(cmd_id_offset) for cmd_class in self._commands}
         self.cmd_by_id = dict((cmd.cmd_id, cmd) for cmd in self.commands)
 
     def send(self, header: bytes, body: bytes) -> None:
         self.peer.send(header, body)
 
+    def send_request(self, request: BaseRequest[_DecodedMsgType]) -> None:
+        command = self.cmd_by_type[request.cmd_type]
+        header, body = command.encode(request.command_payload)
+        self.send(header, body)
+
+    def supports_command(self, cmd_type: Type[Command]) -> bool:
+        return cmd_type in self.cmd_by_type
+
     def __repr__(self) -> str:
         return "(%s, %d)" % (self.name, self.version)
 

tests/p2p/test_peer_collect_sub_proto_msgs.py

@@ -5,10 +5,6 @@
 
 from trinity.protocol.eth.peer import ETHPeer
 from trinity.protocol.eth.commands import GetBlockHeaders, GetNodeData
-from trinity.protocol.eth.requests import (
-    HeaderRequest,
-    NodeDataRequest,
-)
 
 from tests.trinity.core.peer_helpers import (
     get_directly_linked_peers,
@@ -30,11 +26,11 @@ async def test_peer_subscriber_filters_messages(request, event_loop):
 
     with peer.collect_sub_proto_messages() as collector:
         assert collector in peer._subscribers
-        remote.sub_proto.send_get_node_data(NodeDataRequest([b'\x00' * 32]))
-        remote.sub_proto.send_get_block_headers(HeaderRequest(0, 1, 0, False))
-        remote.sub_proto.send_get_node_data(NodeDataRequest([b'\x00' * 32]))
-        remote.sub_proto.send_get_block_headers(HeaderRequest(1, 1, 0, False))
-        remote.sub_proto.send_get_node_data(NodeDataRequest([b'\x00' * 32]))
+        remote.sub_proto.send_get_node_data([b'\x00' * 32])
+        remote.sub_proto.send_get_block_headers(0, 1, 0, False)
+        remote.sub_proto.send_get_node_data([b'\x00' * 32])
+        remote.sub_proto.send_get_block_headers(1, 1, 0, False)
+        remote.sub_proto.send_get_node_data([b'\x00' * 32])
         await asyncio.sleep(0.01)
 
     assert collector not in peer._subscribers
@@ -51,7 +47,7 @@ async def test_peer_subscriber_filters_messages(request, event_loop):
     assert isinstance(all_messages[4][1], GetNodeData)
 
     # make sure it isn't still collecting
-    remote.sub_proto.send_get_block_headers(HeaderRequest(1, 1, 0, False))
+    remote.sub_proto.send_get_block_headers(1, 1, 0, False)
 
     await asyncio.sleep(0.01)
 

tests/p2p/test_peer_subscriber.py

@@ -8,10 +8,6 @@
 
 from trinity.protocol.eth.peer import ETHPeer
 from trinity.protocol.eth.commands import GetBlockHeaders
-from trinity.protocol.eth.requests import (
-    HeaderRequest,
-    NodeDataRequest,
-)
 
 from tests.trinity.core.peer_helpers import (
     get_directly_linked_peers,
@@ -48,11 +44,11 @@ async def test_peer_subscriber_filters_messages(request, event_loop):
     peer.add_subscriber(header_subscriber)
     peer.add_subscriber(all_subscriber)
 
-    remote.sub_proto.send_get_node_data(NodeDataRequest([b'\x00' * 32]))
-    remote.sub_proto.send_get_block_headers(HeaderRequest(0, 1, 0, False))
-    remote.sub_proto.send_get_node_data(NodeDataRequest([b'\x00' * 32]))
-    remote.sub_proto.send_get_block_headers(HeaderRequest(1, 1, 0, False))
-    remote.sub_proto.send_get_node_data(NodeDataRequest([b'\x00' * 32]))
+    remote.sub_proto.send_get_node_data(tuple([b'\x00' * 32]))
+    remote.sub_proto.send_get_block_headers(0, 1, 0, False)
+    remote.sub_proto.send_get_node_data(tuple([b'\x00' * 32]))
+    remote.sub_proto.send_get_block_headers(1, 1, 0, False)
+    remote.sub_proto.send_get_node_data(tuple([b'\x00' * 32]))
 
     # yeild to let remote and peer transmit.
     await asyncio.sleep(0.01)

tests/trinity/core/p2p-proto/test_block_bodies_request_object.py

@@ -19,7 +19,7 @@
 from eth.rlp.transactions import BaseTransactionFields
 
 from trinity.rlp.block_body import BlockBody
-from trinity.protocol.eth.requests import BlockBodiesRequest
+from trinity.protocol.eth.validators import GetBlockBodiesValidator
 
 
 def mk_uncle(block_number):
@@ -76,32 +76,29 @@ def mk_headers(*counts):
 def test_block_bodies_request_empty_response_is_valid():
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, _, _, _, _ = zip(*headers_bundle)
-    request = BlockBodiesRequest(headers)
-    request.validate_response(tuple(), tuple())
+    request = GetBlockBodiesValidator(headers)
+    request.validate_result(tuple())
 
 
 def test_block_bodies_request_valid_with_full_response():
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, bodies, transactions_roots, trie_data_dicts, uncles_hashes = zip(*headers_bundle)
     transactions_bundles = tuple(zip(transactions_roots, trie_data_dicts))
     bodies_bundle = tuple(zip(bodies, transactions_bundles, uncles_hashes))
-    request = BlockBodiesRequest(headers)
-    request.validate_response(bodies, bodies_bundle)
+    request = GetBlockBodiesValidator(headers)
+    request.validate_result(bodies_bundle)
 
 
 def test_block_bodies_request_valid_with_partial_response():
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, bodies, transactions_roots, trie_data_dicts, uncles_hashes = zip(*headers_bundle)
     transactions_bundles = tuple(zip(transactions_roots, trie_data_dicts))
     bodies_bundle = tuple(zip(bodies, transactions_bundles, uncles_hashes))
-    request = BlockBodiesRequest(headers)
+    request = GetBlockBodiesValidator(headers)
 
-    request.validate_response(bodies[:2], bodies_bundle[:2])
-    request.validate_response(bodies[2:], bodies_bundle[2:])
-    request.validate_response(
-        (bodies[0], bodies[2], bodies[3]),
-        (bodies_bundle[0], bodies_bundle[2], bodies_bundle[3]),
-    )
+    request.validate_result(bodies_bundle[:2])
+    request.validate_result(bodies_bundle[2:])
+    request.validate_result((bodies_bundle[0], bodies_bundle[2], bodies_bundle[3]))
 
 
 def test_block_bodies_request_with_fully_invalid_response():
@@ -115,17 +112,17 @@ def test_block_bodies_request_with_fully_invalid_response():
     w_transactions_bundles = tuple(zip(w_transactions_roots, w_trie_data_dicts))
     w_bodies_bundle = tuple(zip(w_bodies, w_transactions_bundles, w_uncles_hashes))
 
-    request = BlockBodiesRequest(headers)
+    request = GetBlockBodiesValidator(headers)
     with pytest.raises(ValidationError):
-        request.validate_response(w_bodies, w_bodies_bundle)
+        request.validate_result(w_bodies_bundle)
 
 
 def test_block_bodies_request_with_extra_unrequested_bodies():
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, bodies, transactions_roots, trie_data_dicts, uncles_hashes = zip(*headers_bundle)
     transactions_bundles = tuple(zip(transactions_roots, trie_data_dicts))
     bodies_bundle = tuple(zip(bodies, transactions_bundles, uncles_hashes))
-    request = BlockBodiesRequest(headers)
+    request = GetBlockBodiesValidator(headers)
 
     wrong_headers_bundle = mk_headers((3, 2), (4, 8), (1, 0), (0, 0))
     w_headers, w_bodies, w_transactions_roots, w_trie_data_dicts, w_uncles_hashes = zip(
@@ -134,9 +131,6 @@ def test_block_bodies_request_with_extra_unrequested_bodies():
     w_transactions_bundles = tuple(zip(w_transactions_roots, w_trie_data_dicts))
     w_bodies_bundle = tuple(zip(w_bodies, w_transactions_bundles, w_uncles_hashes))
 
-    request = BlockBodiesRequest(headers)
+    request = GetBlockBodiesValidator(headers)
     with pytest.raises(ValidationError):
-        request.validate_response(
-            bodies + w_bodies,
-            bodies_bundle + w_bodies_bundle,
-        )
+        request.validate_result(bodies_bundle + w_bodies_bundle)

tests/trinity/core/p2p-proto/test_header_request_object.py renamed to tests/trinity/core/p2p-proto/test_headers_request_validator.py

@@ -5,6 +5,7 @@
 )
 
 from trinity.protocol.common.requests import BaseHeaderRequest
+from trinity.protocol.common.validators import BaseBlockHeadersValidator
 
 
 FORWARD_0_to_5 = (0, 6, 0, False)
@@ -30,8 +31,9 @@ def __init__(self,
         self.skip = skip
         self.reverse = reverse
 
-    def validate_response(self, response):
-        pass
+
+class BlockHeadersValidator(BaseBlockHeadersValidator):
+    protocol_max_request_size = 192
 
 
 @pytest.mark.parametrize(
@@ -127,10 +129,10 @@ def test_header_request_sequence_matching(
         params,
         sequence,
         is_match):
-    request = HeaderRequest(*params)
+    request = BlockHeadersValidator(*params)
 
     if is_match:
-        request.validate_sequence(sequence)
+        request._validate_sequence(sequence)
     else:
         with pytest.raises(ValidationError):
-            request.validate_sequence(sequence)
+            request._validate_sequence(sequence)

tests/trinity/core/p2p-proto/test_node_data_request_object.py

@@ -8,7 +8,7 @@
     ValidationError,
 )
 
-from trinity.protocol.eth.requests import NodeDataRequest
+from trinity.protocol.eth.validators import GetNodeDataValidator
 
 
 def mk_node():
@@ -27,55 +27,51 @@ def mk_node_data(n):
 
 def test_node_data_request_empty_response_is_valid():
     node_keys, _ = mk_node_data(10)
-    request = NodeDataRequest(node_keys)
+    request = GetNodeDataValidator(node_keys)
 
-    request.validate_response(tuple(), tuple())
+    request.validate_result(tuple())
 
 
 def test_node_data_request_with_full_response():
     node_keys, nodes = mk_node_data(10)
-    request = NodeDataRequest(node_keys)
+    request = GetNodeDataValidator(node_keys)
     node_data = tuple(zip(node_keys, nodes))
 
-    request.validate_response(nodes, node_data)
+    request.validate_result(node_data)
 
 
 def test_node_data_request_with_partial_response():
     node_keys, nodes = mk_node_data(10)
-    request = NodeDataRequest(node_keys)
+    request = GetNodeDataValidator(node_keys)
     node_data = tuple(zip(node_keys, nodes))
 
-    request.validate_response(nodes[3:], node_data[3:])
-    request.validate_response(nodes[:3], node_data[:3])
-    request.validate_response(
-        (nodes[1], nodes[8], nodes[4]),
-        (node_data[1], node_data[8], node_data[4]),
-    )
+    request.validate_result(node_data[3:])
+
+    request.validate_result(node_data[:3])
+
+    request.validate_result((node_data[1], node_data[8], node_data[4]))
 
 
 def test_node_data_request_with_fully_invalid_response():
     node_keys, nodes = mk_node_data(10)
-    request = NodeDataRequest(node_keys)
+    request = GetNodeDataValidator(node_keys)
 
     # construct a unique set of other nodes
     other_nodes = tuple(set(mk_node() for _ in range(10)).difference(nodes))
     other_node_data = tuple((keccak(node), node) for node in other_nodes)
 
     with pytest.raises(ValidationError):
-        request.validate_response(other_nodes, other_node_data)
+        request.validate_result(other_node_data)
 
 
 def test_node_data_request_with_extra_unrequested_nodes():
     node_keys, nodes = mk_node_data(10)
-    request = NodeDataRequest(node_keys)
+    request = GetNodeDataValidator(node_keys)
     node_data = tuple(zip(node_keys, nodes))
 
     # construct a unique set of other nodes
     other_nodes = tuple(set(mk_node() for _ in range(10)).difference(nodes))
     other_node_data = tuple((keccak(node), node) for node in other_nodes)
 
     with pytest.raises(ValidationError):
-        request.validate_response(
-            nodes + other_nodes,
-            node_data + other_node_data,
-        )
+        request.validate_result(node_data + other_node_data)

tests/trinity/core/p2p-proto/test_peer_block_body_request_and_response_api.py renamed to tests/trinity/core/p2p-proto/test_peer_block_body_validator_api.py

@@ -6,6 +6,7 @@
 
 from eth.rlp.headers import BlockHeader
 
+import trinity
 from trinity.protocol.eth.peer import ETHPeer
 from trinity.protocol.les.peer import LESPeer
 
@@ -94,11 +95,13 @@ async def send_headers():
 @pytest.mark.asyncio
 async def test_les_peer_get_headers_round_trip(les_peer_and_remote,
                                                params,
+                                               monkeypatch,
                                                headers):
     peer, remote = les_peer_and_remote
+    request_id = 9
+    monkeypatch.setattr(trinity.protocol.les.exchanges, 'gen_request_id', lambda: request_id)
 
     async def send_headers():
-        request_id = peer.requests.get_block_headers.pending_request[0].request_id
         remote.sub_proto.send_block_headers(headers, 0, request_id)
         await asyncio.sleep(0)
 

tests/trinity/core/p2p-proto/test_peer_block_header_request_and_response_api.py renamed to tests/trinity/core/p2p-proto/test_peer_block_header_validator_api.py

@@ -62,8 +62,9 @@ async def test_eth_peer_get_node_data_round_trip(eth_peer_and_remote, node_keys,
     async def send_node_data():
         remote.sub_proto.send_node_data(nodes)
 
+    request = asyncio.ensure_future(peer.requests.get_node_data(node_keys))
     asyncio.ensure_future(send_node_data())
-    response = await peer.requests.get_node_data(node_keys)
+    response = await request
 
     assert len(response) == len(node_keys)
     assert response == node_data