Test buffered valid message

A message can be broken across two buffers, with the split inside its
header.  Usually this will occur when sending many messages, such that
the first buffer fills.

This test uses the RPC to verify that the message is actually being
received in two pieces.

There is a very rare chance of a race condition where the test framework
sends a message in between the two halves of the message under test.  In
this case the peer will almost certainly disconnect and the test will
fail.  An assert has been added to help debugging that rare case.

test/functional/p2p_invalid_messages.py

@@ -3,12 +3,14 @@
 # Distributed under the MIT software license, see the accompanying
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 """Test node responses to invalid network messages."""
+
 from test_framework.messages import (
     CBlockHeader,
     CInv,
     msg_getdata,
     msg_headers,
     msg_inv,
+    msg_ping,
     MSG_TX,
     ser_string,
 )
@@ -17,6 +19,10 @@ from test_framework.mininode import (
     P2PInterface,
 )
 from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import (
+    assert_equal,
+    wait_until,
+)
 
 MSG_LIMIT = 4 * 1000 * 1000  # 4MB, per MAX_PROTOCOL_MESSAGE_LENGTH
 VALID_DATA_LIMIT = MSG_LIMIT - 5  # Account for the 5-byte length prefix
@@ -42,6 +48,7 @@ class InvalidMessagesTest(BitcoinTestFramework):
         self.setup_clean_chain = True
 
     def run_test(self):
+        self.test_buffer()
         self.test_magic_bytes()
         self.test_checksum()
         self.test_size()
@@ -49,6 +56,25 @@ class InvalidMessagesTest(BitcoinTestFramework):
         self.test_large_inv()
         self.test_resource_exhaustion()
 
+    def test_buffer(self):
+        self.log.info("Test message with header split across two buffers, should be received")
+        conn = self.nodes[0].add_p2p_connection(P2PDataStore())
+        # Create valid message
+        msg = conn.build_message(msg_ping(nonce=12345))
+        cut_pos = 12    # Chosen at an arbitrary position within the header
+        # Send message in two pieces
+        before = int(self.nodes[0].getnettotals()['totalbytesrecv'])
+        conn.send_raw_message(msg[:cut_pos])
+        # Wait until node has processed the first half of the message
+        wait_until(lambda: int(self.nodes[0].getnettotals()['totalbytesrecv']) != before)
+        middle = int(self.nodes[0].getnettotals()['totalbytesrecv'])
+        # If this assert fails, we've hit an unlikely race
+        # where the test framework sent a message in between the two halves
+        assert_equal(middle, before + cut_pos)
+        conn.send_raw_message(msg[cut_pos:])
+        conn.sync_with_ping(timeout=1)
+        self.nodes[0].disconnect_p2ps()
+
     def test_magic_bytes(self):
         conn = self.nodes[0].add_p2p_connection(P2PDataStore())
         with self.nodes[0].assert_debug_log(['PROCESSMESSAGE: INVALID MESSAGESTART badmsg']):
@@ -95,13 +121,13 @@ class InvalidMessagesTest(BitcoinTestFramework):
 
     def test_large_inv(self):
         conn = self.nodes[0].add_p2p_connection(P2PInterface())
-        with self.nodes[0].assert_debug_log(['Misbehaving', 'peer=4 (0 -> 20): message inv size() = 50001']):
+        with self.nodes[0].assert_debug_log(['Misbehaving', '(0 -> 20): message inv size() = 50001']):
             msg = msg_inv([CInv(MSG_TX, 1)] * 50001)
             conn.send_and_ping(msg)
-        with self.nodes[0].assert_debug_log(['Misbehaving', 'peer=4 (20 -> 40): message getdata size() = 50001']):
+        with self.nodes[0].assert_debug_log(['Misbehaving', '(20 -> 40): message getdata size() = 50001']):
             msg = msg_getdata([CInv(MSG_TX, 1)] * 50001)
             conn.send_and_ping(msg)
-        with self.nodes[0].assert_debug_log(['Misbehaving', 'peer=4 (40 -> 60): headers message size = 2001']):
+        with self.nodes[0].assert_debug_log(['Misbehaving', '(40 -> 60): headers message size = 2001']):
             msg = msg_headers([CBlockHeader()] * 2001)
             conn.send_and_ping(msg)
         self.nodes[0].disconnect_p2ps()