Merge bitcoin#28287: rpc, test: add sendmsgtopeer rpc and a test fo…

…r net-level deadlock situation

b3a93b4 test: add functional test for deadlock situation (Martin Zumsande)
3557aa4 test: add basic tests for sendmsgtopeer to rpc_net.py (Martin Zumsande)
a9a1d69 rpc: add test-only sendmsgtopeer rpc (Martin Zumsande)

Pull request description:

  This adds a `sendmsgtopeer` rpc (for testing only) that allows a node to send a message (provided in hex) to a peer.
  While we would usually use a `p2p` object instead of a node for this in the test framework, that isn't possible in situations where this message needs to trigger an actual interaction of multiple nodes.

  Use this rpc to add test coverage for the bug fixed in bitcoin#27981 (that just got merged):
  The test lets two nodes (almost) simultaneously send a single large (4MB) p2p message to each other, which would have caused a deadlock previously (making this test fail), but succeeds now.

  As can be seen from the discussion in bitcoin#27981, it was not easy to reproduce this bug without `sendmsgtopeer`. I would imagine that `sendmsgtopeer` could also be helpful in various other test constellations.

ACKs for top commit:
  ajtowns:
    ACK b3a93b4
  sipa:
    ACK b3a93b4
  achow101:
    ACK b3a93b4

Tree-SHA512: 6e22e72402f3c4dd70cddb9e96ea988444720f7a164031df159fbdd48056c8ac77ac53def045d9208a3ca07437c7c8e34f8b4ebc7066c0a84d81cd53f2f4fa5f

src/rpc/client.cpp

@@ -299,6 +299,7 @@ static const CRPCConvertParam vRPCConvertParams[] =
     { "getnodeaddresses", 0, "count"},
     { "addpeeraddress", 1, "port"},
     { "addpeeraddress", 2, "tried"},
+    { "sendmsgtopeer", 0, "peer_id" },
     { "stop", 0, "wait" },
 };
 // clang-format on

src/rpc/net.cpp

@@ -968,6 +968,54 @@ static RPCHelpMan addpeeraddress()
     };
 }
 
+static RPCHelpMan sendmsgtopeer()
+{
+    return RPCHelpMan{
+        "sendmsgtopeer",
+        "Send a p2p message to a peer specified by id.\n"
+        "The message type and body must be provided, the message header will be generated.\n"
+        "This RPC is for testing only.",
+        {
+            {"peer_id", RPCArg::Type::NUM, RPCArg::Optional::NO, "The peer to send the message to."},
+            {"msg_type", RPCArg::Type::STR, RPCArg::Optional::NO, strprintf("The message type (maximum length %i)", CMessageHeader::COMMAND_SIZE)},
+            {"msg", RPCArg::Type::STR_HEX, RPCArg::Optional::NO, "The serialized message body to send, in hex, without a message header"},
+        },
+        RPCResult{RPCResult::Type::OBJ, "", "", std::vector<RPCResult>{}},
+        RPCExamples{
+            HelpExampleCli("sendmsgtopeer", "0 \"addr\" \"ffffff\"") + HelpExampleRpc("sendmsgtopeer", "0 \"addr\" \"ffffff\"")},
+        [&](const RPCHelpMan& self, const JSONRPCRequest& request) -> UniValue {
+            const NodeId peer_id{request.params[0].getInt<int64_t>()};
+            const std::string& msg_type{request.params[1].get_str()};
+            if (msg_type.size() > CMessageHeader::COMMAND_SIZE) {
+                throw JSONRPCError(RPC_INVALID_PARAMETER, strprintf("Error: msg_type too long, max length is %i", CMessageHeader::COMMAND_SIZE));
+            }
+            auto msg{TryParseHex<unsigned char>(request.params[2].get_str())};
+            if (!msg.has_value()) {
+                throw JSONRPCError(RPC_INVALID_PARAMETER, "Error parsing input for msg");
+            }
+
+            NodeContext& node = EnsureAnyNodeContext(request.context);
+            CConnman& connman = EnsureConnman(node);
+
+            CSerializedNetMsg msg_ser;
+            msg_ser.data = msg.value();
+            msg_ser.m_type = msg_type;
+
+            bool success = connman.ForNode(peer_id, [&](CNode* node) {
+                connman.PushMessage(node, std::move(msg_ser));
+                return true;
+            });
+
+            if (!success) {
+                throw JSONRPCError(RPC_MISC_ERROR, "Error: Could not send message to peer");
+            }
+
+            UniValue ret{UniValue::VOBJ};
+            return ret;
+        },
+    };
+}
+
 void RegisterNetRPCCommands(CRPCTable& t)
 {
     static const CRPCCommand commands[]{
@@ -986,6 +1034,7 @@ void RegisterNetRPCCommands(CRPCTable& t)
         {"network", &getnodeaddresses},
         {"hidden", &addconnection},
         {"hidden", &addpeeraddress},
+        {"hidden", &sendmsgtopeer},
     };
     for (const auto& c : commands) {
         t.appendCommand(c.name, &c);

src/test/fuzz/rpc.cpp

@@ -158,6 +158,7 @@ const std::vector<std::string> RPC_COMMANDS_SAFE_FOR_FUZZING{
     "reconsiderblock",
     "scanblocks",
     "scantxoutset",
+    "sendmsgtopeer", // when no peers are connected, no p2p message is sent
     "sendrawtransaction",
     "setmocktime",
     "setnetworkactive",

test/functional/p2p_net_deadlock.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+# Copyright (c) 2023-present The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+import threading
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import random_bytes
+
+
+class NetDeadlockTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.setup_clean_chain = True
+        self.num_nodes = 2
+
+    def run_test(self):
+        node0 = self.nodes[0]
+        node1 = self.nodes[1]
+
+        self.log.info("Simultaneously send a large message on both sides")
+        rand_msg = random_bytes(4000000).hex()
+
+        thread0 = threading.Thread(target=node0.sendmsgtopeer, args=(0, "unknown", rand_msg))
+        thread1 = threading.Thread(target=node1.sendmsgtopeer, args=(0, "unknown", rand_msg))
+
+        thread0.start()
+        thread1.start()
+        thread0.join()
+        thread1.join()
+
+        self.log.info("Check whether a deadlock happened")
+        self.generate(node0, 1)
+        self.sync_blocks()
+
+
+if __name__ == '__main__':
+    NetDeadlockTest().main()

test/functional/rpc_net.py

@@ -65,6 +65,7 @@ def run_test(self):
         self.test_service_flags()
         self.test_getnodeaddresses()
         self.test_addpeeraddress()
+        self.test_sendmsgtopeer()
 
     def test_connection_count(self):
         self.log.info("Test getconnectioncount")
@@ -328,6 +329,37 @@ def test_addpeeraddress(self):
             addrs = node.getnodeaddresses(count=0)  # getnodeaddresses re-runs the addrman checks
             assert_equal(len(addrs), 2)
 
+    def test_sendmsgtopeer(self):
+        node = self.nodes[0]
+
+        self.restart_node(0)
+        self.connect_nodes(0, 1)
+
+        self.log.info("Test sendmsgtopeer")
+        self.log.debug("Send a valid message")
+        with self.nodes[1].assert_debug_log(expected_msgs=["received: addr"]):
+            node.sendmsgtopeer(peer_id=0, msg_type="addr", msg="FFFFFF")
+
+        self.log.debug("Test error for sending to non-existing peer")
+        assert_raises_rpc_error(-1, "Error: Could not send message to peer", node.sendmsgtopeer, peer_id=100, msg_type="addr", msg="FF")
+
+        self.log.debug("Test that zero-length msg_type is allowed")
+        node.sendmsgtopeer(peer_id=0, msg_type="addr", msg="")
+
+        self.log.debug("Test error for msg_type that is too long")
+        assert_raises_rpc_error(-8, "Error: msg_type too long, max length is 12", node.sendmsgtopeer, peer_id=0, msg_type="long_msg_type", msg="FF")
+
+        self.log.debug("Test that unknown msg_type is allowed")
+        node.sendmsgtopeer(peer_id=0, msg_type="unknown", msg="FF")
+
+        self.log.debug("Test that empty msg is allowed")
+        node.sendmsgtopeer(peer_id=0, msg_type="addr", msg="FF")
+
+        self.log.debug("Test that oversized messages are allowed, but get us disconnected")
+        zero_byte_string = b'\x00' * 4000001
+        node.sendmsgtopeer(peer_id=0, msg_type="addr", msg=zero_byte_string.hex())
+        self.wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 0, timeout=10)
+
 
 if __name__ == '__main__':
     NetTest().main()

test/functional/test_runner.py

@@ -268,6 +268,7 @@
     'p2p_leak_tx.py',
     'p2p_eviction.py',
     'p2p_ibd_stalling.py',
+    'p2p_net_deadlock.py',
     'wallet_signmessagewithaddress.py',
     'rpc_signmessagewithprivkey.py',
     'rpc_generate.py',