feat(p2p): Add ping salt and improve rtt information

Author: msbrogli

hathor/p2p/resources/status.py

@@ -69,6 +69,7 @@ def render_GET(self, request):
                 'address': '{}:{}'.format(remote.host, remote.port),
                 'state': conn.state.state_name,
                 # 'received_bytes': conn.received_bytes,
+                'rtt': list(conn.state.rtt_window),
                 'last_message': time.time() - conn.last_message,
                 'plugins': status,
                 'warning_flags': [flag.value for flag in conn.warning_flags],

hathor/p2p/states/ready.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from math import inf
+from collections import deque
 from typing import TYPE_CHECKING, Iterable, Optional
 
 from structlog import get_logger
@@ -53,14 +53,18 @@ def __init__(self, protocol: 'HathorProtocol') -> None:
         # Time we sent last PING message.
         self.ping_start_time: Optional[float] = None
 
+        # Salt used in the last PING message.
+        self.ping_salt: Optional[str] = None
+
+        # Salt size in bytes.
+        self.ping_salt_size: int = 32
+
         # Time we got last PONG response to a PING message.
         self.ping_last_response: float = 0
 
         # Round-trip time of the last PING/PONG.
-        self.ping_rtt: float = inf
-
-        # Minimum round-trip time among PING/PONG.
-        self.ping_min_rtt: float = inf
+        self.rtt_window: deque[float] = deque()
+        self.MAX_RTT_WINDOW: int = 200    # Last 200 samples (~= 10 minutes)
 
         # The last blocks from the best blockchain in the peer
         self.peer_best_blockchain: list[HeightInfo] = []
@@ -146,7 +150,7 @@ def handle_get_peers(self, payload: str) -> None:
             self.send_peers(self.protocol.connections.iter_ready_connections())
 
     def send_peers(self, connections: Iterable['HathorProtocol']) -> None:
-        """ Send a PEERS command with a list of all known peers.
+        """ Send a PEERS command with a list of all connected peers.
         """
         peers = []
         for conn in connections:
@@ -185,28 +189,41 @@ def send_ping(self) -> None:
         """ Send a PING command. Usually you would use `send_ping_if_necessary` to
         prevent wasting bandwidth.
         """
+        # Add a salt number to prevent peers from faking rtt.
         self.ping_start_time = self.reactor.seconds()
-        self.send_message(ProtocolMessages.PING)
+        rng = self.protocol.connections.rng
+        self.ping_salt = rng.randbytes(self.ping_salt_size).hex()
+        self.send_message(ProtocolMessages.PING, self.ping_salt)
 
-    def send_pong(self) -> None:
+    def send_pong(self, salt: str) -> None:
         """ Send a PONG command as a response to a PING command.
         """
-        self.send_message(ProtocolMessages.PONG)
+        self.send_message(ProtocolMessages.PONG, salt)
 
     def handle_ping(self, payload: str) -> None:
         """Executed when a PING command is received. It responds with a PONG message."""
-        self.send_pong()
+        self.send_pong(payload)
 
     def handle_pong(self, payload: str) -> None:
         """Executed when a PONG message is received."""
         if self.ping_start_time is None:
             # This should never happen.
             return
+        if self.ping_salt != payload:
+            # Ignore pong without salts.
+            return
         self.ping_last_response = self.reactor.seconds()
-        self.ping_rtt = self.ping_last_response - self.ping_start_time
-        self.ping_min_rtt = min(self.ping_min_rtt, self.ping_rtt)
+        rtt = self.ping_last_response - self.ping_start_time
+        self.rtt_window.appendleft(rtt)
+        if len(self.rtt_window) > self.MAX_RTT_WINDOW:
+            self.rtt_window.pop()
         self.ping_start_time = None
-        self.log.debug('rtt updated', rtt=self.ping_rtt, min_rtt=self.ping_min_rtt)
+        self.ping_salt = None
+        self.log.debug('rtt updated',
+                       latest=rtt,
+                       min=min(self.rtt_window),
+                       max=max(self.rtt_window),
+                       avg=sum(self.rtt_window) / len(self.rtt_window))
 
     def send_get_best_blockchain(self, n_blocks: Optional[int] = None) -> None:
         """ Send a GET-BEST-BLOCKCHAIN command, requesting a list of the latest

tests/p2p/test_protocol.py

@@ -356,14 +356,14 @@ def test_send_ping(self):
         self.conn.run_one_step()  # TIPS
         self.assertIsConnected()
         self.clock.advance(5)
-        self.assertEqual(b'PING\r\n', self.conn.peek_tr1_value())
-        self.assertEqual(b'PING\r\n', self.conn.peek_tr2_value())
+        self.assertRegex(self.conn.peek_tr1_value(), b'^PING .*\r\n')
+        self.assertRegex(self.conn.peek_tr2_value(), b'^PING .*\r\n')
         self.conn.run_one_step()  # PING
         self.conn.run_one_step()  # GET-TIPS
         self.conn.run_one_step()  # GET-BEST-BLOCKCHAIN
-        self.assertEqual(b'PONG\r\n', self.conn.peek_tr1_value())
-        self.assertEqual(b'PONG\r\n', self.conn.peek_tr2_value())
-        while b'PONG\r\n' in self.conn.peek_tr1_value():
+        self.assertRegex(self.conn.peek_tr1_value(), b'PONG .*\r\n')
+        self.assertRegex(self.conn.peek_tr2_value(), b'PONG .*\r\n')
+        while b'PONG ' in self.conn.peek_tr1_value():
             self.conn.run_one_step()
         self.assertEqual(self.clock.seconds(), self.conn.proto1.last_message)
 
@@ -489,8 +489,8 @@ def test_send_ping(self):
         self.assertAndStepConn(self.conn, b'^TIPS')
         self.assertAndStepConn(self.conn, b'^TIPS')
         self.assertAndStepConn(self.conn, b'^TIPS-END')
-        self.assertEqual(b'PONG\r\n', self.conn.peek_tr1_value())
-        self.assertEqual(b'PONG\r\n', self.conn.peek_tr2_value())
+        self.assertRegex(self.conn.peek_tr1_value(), b'^PONG .*\r\n')
+        self.assertRegex(self.conn.peek_tr2_value(), b'^PONG .*\r\n')
         while b'PONG\r\n' in self.conn.peek_tr1_value():
             self.conn.run_one_step()
         self.assertEqual(self.clock.seconds(), self.conn.proto1.last_message)