Implement round trip API for fetching receipts

Author: pipermerriam

eth/db/trie.py

@@ -1,20 +1,24 @@
 import functools
-from typing import Dict, List, Tuple, Union
+from typing import Dict, Tuple, Union
 
 import rlp
 from trie import (
     HexaryTrie,
 )
 
+from eth_typing import Hash32
+
 from eth.constants import (
     BLANK_ROOT_HASH,
 )
 from eth.rlp.receipts import Receipt
 from eth.rlp.transactions import BaseTransaction
 
+TransactionsOrReceipts = Union[Tuple[Receipt, ...], Tuple[BaseTransaction, ...]]
+TrieRootAndData = Tuple[Hash32, Dict[Hash32, bytes]]
+
 
-def make_trie_root_and_nodes(
-        items: Union[List[Receipt], List[BaseTransaction]]) -> Tuple[bytes, Dict[bytes, bytes]]:
+def make_trie_root_and_nodes(items: TransactionsOrReceipts) -> TrieRootAndData:
     return _make_trie_root_and_nodes(tuple(rlp.encode(item) for item in items))
 
 
@@ -23,8 +27,8 @@ def make_trie_root_and_nodes(
 # as it's common for them to have duplicate receipt_roots. Given that, it probably makes sense to
 # use a relatively small cache size here.
 @functools.lru_cache(128)
-def _make_trie_root_and_nodes(items: Tuple[bytes, ...]) -> Tuple[bytes, Dict[bytes, bytes]]:
-    kv_store = {}  # type: Dict[bytes, bytes]
+def _make_trie_root_and_nodes(items: Tuple[bytes, ...]) -> TrieRootAndData:
+    kv_store = {}  # type: Dict[Hash32, bytes]
     trie = HexaryTrie(kv_store, BLANK_ROOT_HASH)
     with trie.squash_changes() as memory_trie:
         for index, item in enumerate(items):

p2p/peer.py

@@ -1067,8 +1067,8 @@ async def request_stuff() -> None:
         hashes = [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

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

@@ -28,25 +28,28 @@ def test_node_data_request_empty_response_is_valid():
     node_keys, _ = mk_node_data(10)
     request = NodeDataRequest(node_keys)
 
-    request.validate_response(tuple())
+    request.validate_response(tuple(), tuple())
 
 
 def test_node_data_request_with_full_response():
     node_keys, nodes = mk_node_data(10)
     request = NodeDataRequest(node_keys)
     node_data = tuple(zip(node_keys, nodes))
 
-    request.validate_response(node_data)
+    request.validate_response(nodes, node_data)
 
 
 def test_node_data_request_with_partial_response():
     node_keys, nodes = mk_node_data(10)
     request = NodeDataRequest(node_keys)
     node_data = tuple(zip(node_keys, nodes))
 
-    request.validate_response(node_data[3:])
-    request.validate_response(node_data[:3])
-    request.validate_response((node_data[1], node_data[8], node_data[4]))
+    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]),
+    )
 
 
 def test_node_data_request_with_fully_invalid_response():
@@ -58,7 +61,7 @@ def test_node_data_request_with_fully_invalid_response():
     other_node_data = tuple((keccak(node), node) for node in other_nodes)
 
     with pytest.raises(ValidationError):
-        request.validate_response(other_node_data)
+        request.validate_response(other_nodes, other_node_data)
 
 
 def test_node_data_request_with_extra_unrequested_nodes():
@@ -71,4 +74,7 @@ def test_node_data_request_with_extra_unrequested_nodes():
     other_node_data = tuple((keccak(node), node) for node in other_nodes)
 
     with pytest.raises(ValidationError):
-        request.validate_response(node_data + other_node_data)
+        request.validate_response(
+            nodes + other_nodes,
+            node_data + other_node_data,
+        )

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

@@ -0,0 +1,143 @@
+import asyncio
+import os
+import time
+
+import pytest
+
+from eth_utils import to_tuple
+
+from eth.db.trie import make_trie_root_and_nodes
+from eth.rlp.headers import BlockHeader
+from eth.rlp.receipts import Receipt
+
+from trinity.protocol.eth.peer import ETHPeer
+
+from tests.trinity.core.peer_helpers import (
+    get_directly_linked_peers,
+)
+
+
+@pytest.fixture
+async def eth_peer_and_remote(request, event_loop):
+    peer, remote = await get_directly_linked_peers(
+        request,
+        event_loop,
+        peer1_class=ETHPeer,
+        peer2_class=ETHPeer,
+    )
+    return peer, remote
+
+
+@to_tuple
+def mk_receipts(num_receipts):
+    for _ in range(num_receipts):
+        yield Receipt(
+            state_root=os.urandom(32),
+            gas_used=21000,
+            bloom=0,
+            logs=[],
+        )
+
+
+def mk_header_and_receipts(block_number, num_receipts):
+    receipts = mk_receipts(num_receipts)
+    root_hash, trie_root_and_data = make_trie_root_and_nodes(receipts)
+    header = BlockHeader(
+        difficulty=1000000,
+        block_number=block_number,
+        gas_limit=3141592,
+        timestamp=int(time.time()),
+        receipt_root=root_hash,
+    )
+    return header, receipts, (root_hash, trie_root_and_data)
+
+
+@to_tuple
+def mk_headers(*counts):
+    for idx, num_receipts in enumerate(counts, 1):
+        yield mk_header_and_receipts(idx, num_receipts)
+
+
+@pytest.mark.asyncio
+async def test_eth_peer_get_receipts_round_trip_with_full_response(eth_peer_and_remote):
+    peer, remote = eth_peer_and_remote
+
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, receipts, trie_roots_and_data = zip(*headers_bundle)
+    receipts_bundle = tuple(zip(receipts, trie_roots_and_data))
+
+    async def send_receipts():
+        remote.sub_proto.send_receipts(receipts)
+        await asyncio.sleep(0)
+
+    asyncio.ensure_future(send_receipts())
+    response = await peer.requests.get_receipts(headers)
+
+    assert len(response) == len(headers)
+    assert response == receipts_bundle
+
+
+@pytest.mark.asyncio
+async def test_eth_peer_get_receipts_round_trip_with_partial_response(eth_peer_and_remote):
+    peer, remote = eth_peer_and_remote
+
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, receipts, trie_roots_and_data = zip(*headers_bundle)
+    receipts_bundle = tuple(zip(receipts, trie_roots_and_data))
+
+    async def send_receipts():
+        remote.sub_proto.send_receipts((receipts[2], receipts[1], receipts[4]))
+        await asyncio.sleep(0)
+
+    asyncio.ensure_future(send_receipts())
+    response = await peer.requests.get_receipts(headers)
+
+    assert len(response) == 3
+    assert response == (receipts_bundle[2], receipts_bundle[1], receipts_bundle[4])
+
+
+@pytest.mark.asyncio
+async def test_eth_peer_get_receipts_round_trip_with_noise(eth_peer_and_remote):
+    peer, remote = eth_peer_and_remote
+
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, receipts, trie_roots_and_data = zip(*headers_bundle)
+    receipts_bundle = tuple(zip(receipts, trie_roots_and_data))
+
+    async def send_receipts():
+        remote.sub_proto.send_transactions([])
+        await asyncio.sleep(0)
+        remote.sub_proto.send_receipts(receipts)
+        await asyncio.sleep(0)
+        remote.sub_proto.send_transactions([])
+        await asyncio.sleep(0)
+
+    asyncio.ensure_future(send_receipts())
+    response = await peer.requests.get_receipts(headers)
+
+    assert len(response) == len(headers)
+    assert response == receipts_bundle
+
+
+@pytest.mark.asyncio
+async def test_eth_peer_get_receipts_round_trip_no_match_invalid_response(eth_peer_and_remote):
+    peer, remote = eth_peer_and_remote
+
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, receipts, trie_roots_and_data = zip(*headers_bundle)
+    receipts_bundle = tuple(zip(receipts, trie_roots_and_data))
+
+    wrong_headers = mk_headers(4, 3, 8)
+    _, wrong_receipts, _ = zip(*wrong_headers)
+
+    async def send_receipts():
+        remote.sub_proto.send_receipts(wrong_receipts)
+        await asyncio.sleep(0)
+        remote.sub_proto.send_receipts(receipts)
+        await asyncio.sleep(0)
+
+    asyncio.ensure_future(send_receipts())
+    response = await peer.requests.get_receipts(headers)
+
+    assert len(response) == len(headers)
+    assert response == receipts_bundle

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

@@ -0,0 +1,105 @@
+import os
+import time
+
+import pytest
+
+from eth_utils import to_tuple
+
+from eth.db.trie import make_trie_root_and_nodes
+from eth.rlp.headers import BlockHeader
+from eth.rlp.receipts import Receipt
+
+from p2p.exceptions import ValidationError
+
+from trinity.protocol.eth.requests import ReceiptsRequest
+
+
+@to_tuple
+def mk_receipts(num_receipts):
+    for _ in range(num_receipts):
+        yield Receipt(
+            state_root=os.urandom(32),
+            gas_used=21000,
+            bloom=0,
+            logs=[],
+        )
+
+
+def mk_header_and_receipts(block_number, num_receipts):
+    receipts = mk_receipts(num_receipts)
+    root_hash, trie_root_and_data = make_trie_root_and_nodes(receipts)
+    header = BlockHeader(
+        difficulty=1000000,
+        block_number=block_number,
+        gas_limit=3141592,
+        timestamp=int(time.time()),
+        receipt_root=root_hash,
+    )
+    return header, receipts, (root_hash, trie_root_and_data)
+
+
+@to_tuple
+def mk_headers(*counts):
+    for idx, num_receipts in enumerate(counts, 1):
+        yield mk_header_and_receipts(idx, num_receipts)
+
+
+def test_receipts_request_empty_response_is_valid():
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, _, _ = zip(*headers_bundle)
+    request = ReceiptsRequest(headers)
+    request.validate_response(tuple(), tuple())
+
+
+def test_receipts_request_valid_with_full_response():
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, receipts, trie_roots_and_data = zip(*headers_bundle)
+    receipts_bundle = tuple(zip(receipts, trie_roots_and_data))
+    request = ReceiptsRequest(headers)
+    request.validate_response(receipts, receipts_bundle)
+
+
+def test_receipts_request_valid_with_partial_response():
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, receipts, trie_roots_and_data = zip(*headers_bundle)
+    receipts_bundle = tuple(zip(receipts, trie_roots_and_data))
+    request = ReceiptsRequest(headers)
+
+    request.validate_response(receipts[:3], receipts_bundle[:3])
+    request.validate_response(receipts[2:], receipts_bundle[2:])
+    request.validate_response(
+        (receipts[1], receipts[3], receipts[4]),
+        (receipts_bundle[1], receipts_bundle[3], receipts_bundle[4]),
+    )
+
+
+def test_receipts_request_with_fully_invalid_response():
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, _, _ = zip(*headers_bundle)
+
+    wrong_headers = mk_headers(4, 3, 8)
+    _, wrong_receipts, wrong_trie_roots_and_data = zip(*wrong_headers)
+    receipts_bundle = tuple(zip(wrong_receipts, wrong_trie_roots_and_data))
+
+    request = ReceiptsRequest(headers)
+
+    with pytest.raises(ValidationError):
+        request.validate_response(wrong_receipts, receipts_bundle)
+
+
+def test_receipts_request_with_extra_unrequested_receipts():
+    headers_bundle = mk_headers(1, 3, 2, 5, 4)
+    headers, receipts, trie_roots_and_data = zip(*headers_bundle)
+    receipts_bundle = tuple(zip(receipts, trie_roots_and_data))
+
+    wrong_headers = mk_headers(4, 3, 8)
+    _, wrong_receipts, wrong_trie_roots_and_data = zip(*wrong_headers)
+    extra_receipts_bundle = tuple(zip(wrong_receipts, wrong_trie_roots_and_data))
+
+    request = ReceiptsRequest(headers)
+
+    with pytest.raises(ValidationError):
+        request.validate_response(
+            receipts + wrong_receipts,
+            receipts_bundle + extra_receipts_bundle,
+        )

trinity/p2p/handlers.py

@@ -1,4 +1,5 @@
 from typing import (
+    Any,
     AsyncGenerator,
     List,
     Tuple,
@@ -80,7 +81,7 @@ async def handle_get_node_data(self, peer: ETHPeer, node_hashes: List[Hash32]) -
         peer.sub_proto.send_node_data(tuple(nodes))
 
     async def lookup_headers(self,
-                             request: BaseHeaderRequest) -> Tuple[BlockHeader, ...]:
+                             request: BaseHeaderRequest[Any]) -> Tuple[BlockHeader, ...]:
         """
         Lookup :max_headers: headers starting at :block_number_or_hash:, skipping :skip: items
         between each, in reverse order if :reverse: is True.
@@ -101,7 +102,7 @@ async def lookup_headers(self,
         return headers
 
     async def _get_block_numbers_for_request(self,
-                                             request: BaseHeaderRequest
+                                             request: BaseHeaderRequest[Any],
                                              ) -> Tuple[BlockNumber, ...]:
         """
         Generate the block numbers for a given `HeaderRequest`.