Merge 93450294d88261ab82dc845971ae22b89bc2cf62 into 5f4422d68dc3530c353af1f87499de1c864b60ad

3301d2cbe8c3b76c97285d75fa59637cb6952d0b qa: Wait for txindex to avoid race condition (Hodlinator)
9bfb0d75ba10591cc6c9620f9fd1ecc0e55e7a48 qa: Remove unnecessary -txindex args (Hodlinator)
7ac281c19cd3d11f316dbbb3308eabf1ad4f26d6 qa: Add missing coverage of corrupt indexes (Hodlinator)

Pull request description:

  - Add synchronization in 3 places where if the Transaction Index happens to be slow, we get rare test failures when querying it for transactions (one such case experienced on Windows, prompting investigation).
  - Remove unnecessary TxIndex initialization in some tests.
  - Add some test coverage where TxIndex aspect could be tested in feature_init.py.

ACKs for top commit:
  fjahr:
    re-ACK 3301d2cbe8c3b76c97285d75fa59637cb6952d0b
  mzumsande:
    Code Review ACK 3301d2cbe8c3b76c97285d75fa59637cb6952d0b
  furszy:
    Code review ACK 3301d2cbe8c3b76c97285d75fa59637cb6952d0b
  Prabhat1308:
    Concept ACK [`3301d2c`](3301d2cbe8)

Tree-SHA512: 7c2019e38455f344856aaf6b381faafbd88d53dc88d13309deb718c1dcfbee4ccca7c7f1b66917395503a6f94c3b216a007ad432cc8b93d0309db9805f38d602

src/CMakeLists.txt

@@ -145,6 +145,7 @@ add_library(bitcoin_common STATIC EXCLUDE_FROM_ALL
   common/run_command.cpp
   common/settings.cpp
   common/signmessage.cpp
+  common/sockman.cpp
   common/system.cpp
   common/url.cpp
   compressor.cpp
@@ -152,6 +153,7 @@ add_library(bitcoin_common STATIC EXCLUDE_FROM_ALL
   core_write.cpp
   deploymentinfo.cpp
   external_signer.cpp
+  i2p.cpp
   init/common.cpp
   kernel/chainparams.cpp
   key.cpp
@@ -229,7 +231,6 @@ add_library(bitcoin_node STATIC EXCLUDE_FROM_ALL
   headerssync.cpp
   httprpc.cpp
   httpserver.cpp
-  i2p.cpp
   index/base.cpp
   index/blockfilterindex.cpp
   index/coinstatsindex.cpp

src/common/sockman.cpp

@@ -0,0 +1,535 @@
+// Copyright (c) 2024-present The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or https://opensource.org/license/mit/.
+
+#include <bitcoin-build-config.h> // IWYU pragma: keep
+
+#include <common/sockman.h>
+#include <logging.h>
+#include <netbase.h>
+#include <util/sock.h>
+#include <util/thread.h>
+
+#include <cassert>
+
+// The set of sockets cannot be modified while waiting
+// The sleep time needs to be small to avoid new sockets stalling
+static constexpr auto SELECT_TIMEOUT{50ms};
+
+/** Get the bind address for a socket as CService. */
+static CService GetBindAddress(const Sock& sock)
+{
+    CService addr_bind;
+    struct sockaddr_storage sockaddr_bind;
+    socklen_t sockaddr_bind_len = sizeof(sockaddr_bind);
+    if (!sock.GetSockName((struct sockaddr*)&sockaddr_bind, &sockaddr_bind_len)) {
+        addr_bind.SetSockAddr((const struct sockaddr*)&sockaddr_bind, sockaddr_bind_len);
+    } else {
+        LogPrintLevel(BCLog::NET, BCLog::Level::Warning, "getsockname failed\n");
+    }
+    return addr_bind;
+}
+
+bool SockMan::BindAndStartListening(const CService& to, bilingual_str& err_msg)
+{
+    // Create socket for listening for incoming connections
+    sockaddr_storage storage;
+    socklen_t len{sizeof(storage)};
+    if (!to.GetSockAddr(reinterpret_cast<sockaddr*>(&storage), &len)) {
+        err_msg = Untranslated(strprintf("Bind address family for %s not supported", to.ToStringAddrPort()));
+        return false;
+    }
+
+    std::unique_ptr<Sock> sock{CreateSock(to.GetSAFamily(), SOCK_STREAM, IPPROTO_TCP)};
+    if (!sock) {
+        err_msg = Untranslated(strprintf("Cannot create %s listen socket: %s",
+                                         to.ToStringAddrPort(),
+                                         NetworkErrorString(WSAGetLastError())));
+        return false;
+    }
+
+    int one{1};
+
+    // Allow binding if the port is still in TIME_WAIT state after
+    // the program was closed and restarted.
+    if (sock->SetSockOpt(SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<sockopt_arg_type>(&one), sizeof(one)) == SOCKET_ERROR) {
+        LogPrintLevel(BCLog::NET,
+                      BCLog::Level::Info,
+                      "Cannot set SO_REUSEADDR on %s listen socket: %s, continuing anyway\n",
+                      to.ToStringAddrPort(),
+                      NetworkErrorString(WSAGetLastError()));
+    }
+
+    // some systems don't have IPV6_V6ONLY but are always v6only; others do have the option
+    // and enable it by default or not. Try to enable it, if possible.
+    if (to.IsIPv6()) {
+#ifdef IPV6_V6ONLY
+        if (sock->SetSockOpt(IPPROTO_IPV6, IPV6_V6ONLY, reinterpret_cast<sockopt_arg_type>(&one), sizeof(one)) == SOCKET_ERROR) {
+            LogPrintLevel(BCLog::NET,
+                          BCLog::Level::Info,
+                          "Cannot set IPV6_V6ONLY on %s listen socket: %s, continuing anyway\n",
+                          to.ToStringAddrPort(),
+                          NetworkErrorString(WSAGetLastError()));
+        }
+#endif
+#ifdef WIN32
+        int prot_level{PROTECTION_LEVEL_UNRESTRICTED};
+        if (sock->SetSockOpt(IPPROTO_IPV6,
+                             IPV6_PROTECTION_LEVEL,
+                             reinterpret_cast<const char*>(&prot_level),
+                             sizeof(prot_level)) == SOCKET_ERROR) {
+            LogPrintLevel(BCLog::NET,
+                          BCLog::Level::Info,
+                          "Cannot set IPV6_PROTECTION_LEVEL on %s listen socket: %s, continuing anyway\n",
+                          to.ToStringAddrPort(),
+                          NetworkErrorString(WSAGetLastError()));
+        }
+#endif
+    }
+
+    if (sock->Bind(reinterpret_cast<sockaddr*>(&storage), len) == SOCKET_ERROR) {
+        const int err{WSAGetLastError()};
+        if (err == WSAEADDRINUSE) {
+            err_msg = strprintf(_("Unable to bind to %s on this computer. %s is probably already running."),
+                                to.ToStringAddrPort(),
+                                CLIENT_NAME);
+        } else {
+            err_msg = strprintf(_("Unable to bind to %s on this computer (bind returned error %s)"),
+                                to.ToStringAddrPort(),
+                                NetworkErrorString(err));
+        }
+        return false;
+    }
+
+    // Listen for incoming connections
+    if (sock->Listen(SOMAXCONN) == SOCKET_ERROR) {
+        err_msg = strprintf(_("Cannot listen on %s: %s"), to.ToStringAddrPort(), NetworkErrorString(WSAGetLastError()));
+        return false;
+    }
+
+    m_listen.emplace_back(std::move(sock));
+
+    return true;
+}
+
+void SockMan::StartSocketsThreads(const Options& options)
+{
+    m_thread_socket_handler = std::thread(
+        &util::TraceThread, options.socket_handler_thread_name, [this] { ThreadSocketHandler(); });
+
+    if (options.i2p.has_value()) {
+        m_i2p_sam_session = std::make_unique<i2p::sam::Session>(
+            options.i2p->private_key_file, options.i2p->sam_proxy, &interruptNet);
+
+        m_thread_i2p_accept =
+            std::thread(&util::TraceThread, options.i2p->accept_thread_name, [this] { ThreadI2PAccept(); });
+    }
+}
+
+void SockMan::JoinSocketsThreads()
+{
+    if (m_thread_i2p_accept.joinable()) {
+        m_thread_i2p_accept.join();
+    }
+
+    if (m_thread_socket_handler.joinable()) {
+        m_thread_socket_handler.join();
+    }
+}
+
+std::optional<SockMan::Id>
+SockMan::ConnectAndMakeId(const std::variant<CService, StringHostIntPort>& to,
+                          bool is_important,
+                          std::optional<Proxy> proxy,
+                          bool& proxy_failed,
+                          CService& me)
+{
+    AssertLockNotHeld(m_connected_mutex);
+    AssertLockNotHeld(m_unused_i2p_sessions_mutex);
+
+    std::unique_ptr<Sock> sock;
+    std::unique_ptr<i2p::sam::Session> i2p_transient_session;
+
+    Assume(!me.IsValid());
+
+    if (std::holds_alternative<CService>(to)) {
+        const CService& addr_to{std::get<CService>(to)};
+        if (addr_to.IsI2P()) {
+            if (!Assume(proxy.has_value())) {
+                return std::nullopt;
+            }
+
+            i2p::Connection conn;
+            bool connected{false};
+
+            if (m_i2p_sam_session) {
+                connected = m_i2p_sam_session->Connect(addr_to, conn, proxy_failed);
+            } else {
+                {
+                    LOCK(m_unused_i2p_sessions_mutex);
+                    if (m_unused_i2p_sessions.empty()) {
+                        i2p_transient_session = std::make_unique<i2p::sam::Session>(proxy.value(), &interruptNet);
+                    } else {
+                        i2p_transient_session.swap(m_unused_i2p_sessions.front());
+                        m_unused_i2p_sessions.pop();
+                    }
+                }
+                connected = i2p_transient_session->Connect(addr_to, conn, proxy_failed);
+                if (!connected) {
+                    LOCK(m_unused_i2p_sessions_mutex);
+                    if (m_unused_i2p_sessions.size() < MAX_UNUSED_I2P_SESSIONS_SIZE) {
+                        m_unused_i2p_sessions.emplace(i2p_transient_session.release());
+                    }
+                }
+            }
+
+            if (connected) {
+                sock = std::move(conn.sock);
+                me = conn.me;
+            }
+        } else if (proxy.has_value()) {
+            sock = ConnectThroughProxy(proxy.value(), addr_to.ToStringAddr(), addr_to.GetPort(), proxy_failed);
+        } else {
+            sock = ConnectDirectly(addr_to, is_important);
+        }
+    } else {
+        if (!Assume(proxy.has_value())) {
+            return std::nullopt;
+        }
+
+        const auto& hostport{std::get<StringHostIntPort>(to)};
+
+        bool dummy_proxy_failed;
+        sock = ConnectThroughProxy(proxy.value(), hostport.host, hostport.port, dummy_proxy_failed);
+    }
+
+    if (!sock) {
+        return std::nullopt;
+    }
+
+    if (!me.IsValid()) {
+        me = GetBindAddress(*sock);
+    }
+
+    const Id id{GetNewId()};
+
+    {
+        LOCK(m_connected_mutex);
+        m_connected.emplace(id, std::make_shared<ConnectionSockets>(std::move(sock),
+                                                                    std::move(i2p_transient_session)));
+    }
+
+    return id;
+}
+
+bool SockMan::CloseConnection(Id id)
+{
+    LOCK(m_connected_mutex);
+    return m_connected.erase(id) > 0;
+}
+
+ssize_t SockMan::SendBytes(Id id,
+                           std::span<const unsigned char> data,
+                           bool will_send_more,
+                           std::string& errmsg) const
+{
+    AssertLockNotHeld(m_connected_mutex);
+
+    if (data.empty()) {
+        return 0;
+    }
+
+    auto sockets{GetConnectionSockets(id)};
+    if (!sockets) {
+        // Bail out immediately and just leave things in the caller's send queue.
+        return 0;
+    }
+
+    int flags{MSG_NOSIGNAL | MSG_DONTWAIT};
+#ifdef MSG_MORE
+    if (will_send_more) {
+        flags |= MSG_MORE;
+    }
+#endif
+
+    const ssize_t sent{WITH_LOCK(
+        sockets->mutex,
+        return sockets->sock->Send(reinterpret_cast<const char*>(data.data()), data.size(), flags);)};
+
+    if (sent >= 0) {
+        return sent;
+    }
+
+    const int err{WSAGetLastError()};
+    if (err == WSAEWOULDBLOCK || err == WSAEMSGSIZE || err == WSAEINTR || err == WSAEINPROGRESS) {
+        return 0;
+    }
+    errmsg = NetworkErrorString(err);
+    return -1;
+}
+
+void SockMan::StopListening()
+{
+    m_listen.clear();
+}
+
+bool SockMan::ShouldTryToSend(Id id) const { return true; }
+
+bool SockMan::ShouldTryToRecv(Id id) const { return true; }
+
+void SockMan::EventIOLoopCompletedForOne(Id id) {}
+
+void SockMan::EventIOLoopCompletedForAll() {}
+
+void SockMan::EventI2PStatus(const CService&, I2PStatus) {}
+
+void SockMan::TestOnlyAddExistentConnection(Id id, std::unique_ptr<Sock>&& sock)
+{
+    LOCK(m_connected_mutex);
+    const auto result{m_connected.emplace(id, std::make_shared<ConnectionSockets>(std::move(sock)))};
+    assert(result.second);
+}
+
+void SockMan::ThreadI2PAccept()
+{
+    AssertLockNotHeld(m_connected_mutex);
+
+    static constexpr auto err_wait_begin = 1s;
+    static constexpr auto err_wait_cap = 5min;
+    auto err_wait = err_wait_begin;
+
+    i2p::Connection conn;
+
+    auto SleepOnFailure = [&]() {
+        interruptNet.sleep_for(err_wait);
+        if (err_wait < err_wait_cap) {
+            err_wait += 1s;
+        }
+    };
+
+    while (!interruptNet) {
+
+        if (!m_i2p_sam_session->Listen(conn)) {
+            EventI2PStatus(conn.me, SockMan::I2PStatus::STOP_LISTENING);
+            SleepOnFailure();
+            continue;
+        }
+
+        EventI2PStatus(conn.me, SockMan::I2PStatus::START_LISTENING);
+
+        if (!m_i2p_sam_session->Accept(conn)) {
+            SleepOnFailure();
+            continue;
+        }
+
+        Assume(conn.me.IsI2P());
+        Assume(conn.peer.IsI2P());
+
+        NewSockAccepted(std::move(conn.sock), conn.me, conn.peer);
+
+        err_wait = err_wait_begin;
+    }
+}
+
+void SockMan::ThreadSocketHandler()
+{
+    AssertLockNotHeld(m_connected_mutex);
+
+    while (!interruptNet) {
+        EventIOLoopCompletedForAll();
+
+        // Check for the readiness of the already connected sockets and the
+        // listening sockets in one call ("readiness" as in poll(2) or
+        // select(2)). If none are ready, wait for a short while and return
+        // empty sets.
+        auto io_readiness{GenerateWaitSockets()};
+        if (io_readiness.events_per_sock.empty() ||
+            // WaitMany() may as well be a static method, the context of the first Sock in the vector is not relevant.
+            !io_readiness.events_per_sock.begin()->first->WaitMany(SELECT_TIMEOUT,
+                                                                   io_readiness.events_per_sock)) {
+            interruptNet.sleep_for(SELECT_TIMEOUT);
+        }
+
+        // Service (send/receive) each of the already connected sockets.
+        SocketHandlerConnected(io_readiness);
+
+        // Accept new connections from listening sockets.
+        SocketHandlerListening(io_readiness.events_per_sock);
+    }
+}
+
+std::unique_ptr<Sock> SockMan::AcceptConnection(const Sock& listen_sock, CService& addr)
+{
+    sockaddr_storage storage;
+    socklen_t len{sizeof(storage)};
+
+    auto sock{listen_sock.Accept(reinterpret_cast<sockaddr*>(&storage), &len)};
+
+    if (!sock) {
+        const int err{WSAGetLastError()};
+        if (err != WSAEWOULDBLOCK) {
+            LogPrintLevel(BCLog::NET,
+                          BCLog::Level::Error,
+                          "Cannot accept new connection: %s\n",
+                          NetworkErrorString(err));
+        }
+        return {};
+    }
+
+    if (!addr.SetSockAddr(reinterpret_cast<sockaddr*>(&storage), len)) {
+        LogPrintLevel(BCLog::NET, BCLog::Level::Warning, "Unknown socket family\n");
+    }
+
+    return sock;
+}
+
+void SockMan::NewSockAccepted(std::unique_ptr<Sock>&& sock, const CService& me, const CService& them)
+{
+    AssertLockNotHeld(m_connected_mutex);
+
+    if (!sock->IsSelectable()) {
+        LogPrintf("connection from %s dropped: non-selectable socket\n", them.ToStringAddrPort());
+        return;
+    }
+
+    // According to the internet TCP_NODELAY is not carried into accepted sockets
+    // on all platforms.  Set it again here just to be sure.
+    const int on{1};
+    if (sock->SetSockOpt(IPPROTO_TCP, TCP_NODELAY, &on, sizeof(on)) == SOCKET_ERROR) {
+        LogDebug(BCLog::NET, "connection from %s: unable to set TCP_NODELAY, continuing anyway\n",
+                 them.ToStringAddrPort());
+    }
+
+    const Id id{GetNewId()};
+
+    {
+        LOCK(m_connected_mutex);
+        m_connected.emplace(id, std::make_shared<ConnectionSockets>(std::move(sock)));
+    }
+
+    if (!EventNewConnectionAccepted(id, me, them)) {
+        CloseConnection(id);
+    }
+}
+
+SockMan::Id SockMan::GetNewId()
+{
+    return m_next_id.fetch_add(1, std::memory_order_relaxed);
+}
+
+SockMan::IOReadiness SockMan::GenerateWaitSockets()
+{
+    AssertLockNotHeld(m_connected_mutex);
+
+    IOReadiness io_readiness;
+
+    for (const auto& sock : m_listen) {
+        io_readiness.events_per_sock.emplace(sock, Sock::Events{Sock::RECV});
+    }
+
+    auto connected_snapshot{WITH_LOCK(m_connected_mutex, return m_connected;)};
+
+    for (const auto& [id, sockets] : connected_snapshot) {
+        const bool select_recv{ShouldTryToRecv(id)};
+        const bool select_send{ShouldTryToSend(id)};
+        if (!select_recv && !select_send) continue;
+
+        Sock::Event event = (select_send ? Sock::SEND : 0) | (select_recv ? Sock::RECV : 0);
+        io_readiness.events_per_sock.emplace(sockets->sock, Sock::Events{event});
+        io_readiness.ids_per_sock.emplace(sockets->sock, id);
+    }
+
+    return io_readiness;
+}
+
+void SockMan::SocketHandlerConnected(const IOReadiness& io_readiness)
+{
+    AssertLockNotHeld(m_connected_mutex);
+
+    for (const auto& [sock, events] : io_readiness.events_per_sock) {
+        if (interruptNet) {
+            return;
+        }
+
+        auto it{io_readiness.ids_per_sock.find(sock)};
+        if (it == io_readiness.ids_per_sock.end()) {
+            continue;
+        }
+        const Id id{it->second};
+
+        bool send_ready = events.occurred & Sock::SEND; // Sock::SEND could only be set if ShouldTryToSend() has returned true in GenerateWaitSockets().
+        bool recv_ready = events.occurred & Sock::RECV; // Sock::RECV could only be set if ShouldTryToRecv() has returned true in GenerateWaitSockets().
+        bool err_ready = events.occurred & Sock::ERR;
+
+        if (send_ready) {
+            bool cancel_recv;
+
+            EventReadyToSend(id, cancel_recv);
+
+            if (cancel_recv) {
+                recv_ready = false;
+            }
+        }
+
+        if (recv_ready || err_ready) {
+            uint8_t buf[0x10000]; // typical socket buffer is 8K-64K
+
+            auto sockets{GetConnectionSockets(id)};
+            if (!sockets) {
+                continue;
+            }
+
+            const ssize_t nrecv{WITH_LOCK(
+                sockets->mutex,
+                return sockets->sock->Recv(buf, sizeof(buf), MSG_DONTWAIT);)};
+
+            if (nrecv < 0) { // In all cases (including -1 and 0) EventIOLoopCompletedForOne() should be executed after this, don't change the code to skip it.
+                const int err = WSAGetLastError();
+                if (err != WSAEWOULDBLOCK && err != WSAEMSGSIZE && err != WSAEINTR && err != WSAEINPROGRESS) {
+                    EventGotPermanentReadError(id, NetworkErrorString(err));
+                }
+            } else if (nrecv == 0) {
+                EventGotEOF(id);
+            } else {
+                EventGotData(id, {buf, static_cast<size_t>(nrecv)});
+            }
+        }
+
+        EventIOLoopCompletedForOne(id);
+    }
+}
+
+void SockMan::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock)
+{
+    AssertLockNotHeld(m_connected_mutex);
+
+    for (const auto& sock : m_listen) {
+        if (interruptNet) {
+            return;
+        }
+        const auto it = events_per_sock.find(sock);
+        if (it != events_per_sock.end() && it->second.occurred & Sock::RECV) {
+            CService addr_accepted;
+
+            auto sock_accepted{AcceptConnection(*sock, addr_accepted)};
+
+            if (sock_accepted) {
+                NewSockAccepted(std::move(sock_accepted), GetBindAddress(*sock), addr_accepted);
+            }
+        }
+    }
+}
+
+std::shared_ptr<SockMan::ConnectionSockets> SockMan::GetConnectionSockets(Id id) const
+{
+    LOCK(m_connected_mutex);
+
+    auto it{m_connected.find(id)};
+    if (it == m_connected.end()) {
+        // There is no socket in case we've already disconnected, or in test cases without
+        // real connections.
+        return {};
+    }
+
+    return it->second;
+}

src/common/sockman.h

@@ -0,0 +1,463 @@
+// Copyright (c) 2024-present The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or https://opensource.org/license/mit/.
+
+#ifndef BITCOIN_COMMON_SOCKMAN_H
+#define BITCOIN_COMMON_SOCKMAN_H
+
+#include <i2p.h>
+#include <netaddress.h>
+#include <netbase.h>
+#include <util/fs.h>
+#include <util/sock.h>
+#include <util/translation.h>
+
+#include <atomic>
+#include <memory>
+#include <optional>
+#include <queue>
+#include <span>
+#include <thread>
+#include <variant>
+#include <vector>
+
+/**
+ * A socket manager class which handles socket operations.
+ * To use this class, inherit from it and implement the pure virtual methods.
+ * Handled operations:
+ * - binding and listening on sockets
+ * - starting of necessary threads to process socket operations
+ * - accepting incoming connections
+ * - making outbound connections
+ * - closing connections
+ * - waiting for IO readiness on sockets and doing send/recv accordingly
+ */
+class SockMan
+{
+public:
+    /**
+     * Each connection is assigned an unique id of this type.
+     */
+    using Id = int64_t;
+
+    /**
+     * Possible status changes that can be passed to `EventI2PStatus()`.
+     */
+    enum class I2PStatus : uint8_t {
+        /// The listen succeeded and we are now listening for incoming I2P connections.
+        START_LISTENING,
+
+        /// The listen failed and now we are not listening (even if START_LISTENING was signaled before).
+        STOP_LISTENING,
+    };
+
+    virtual ~SockMan() = default;
+
+    //
+    // Non-virtual functions, to be reused by children classes.
+    //
+
+    /**
+     * Bind to a new address:port, start listening and add the listen socket to `m_listen`.
+     * Should be called before `StartSocketsThreads()`.
+     * @param[in] to Where to bind.
+     * @param[out] err_msg Error string if an error occurs.
+     * @retval true Success.
+     * @retval false Failure, `err_msg` will be set.
+     */
+    bool BindAndStartListening(const CService& to, bilingual_str& err_msg);
+
+    /**
+     * Options to influence `StartSocketsThreads()`.
+     */
+    struct Options {
+        std::string_view socket_handler_thread_name;
+
+        struct I2P {
+            explicit I2P(const fs::path& file, const Proxy& proxy, std::string_view accept_thread_name)
+                : private_key_file{file},
+                  sam_proxy{proxy},
+                  accept_thread_name{accept_thread_name}
+            {}
+
+            const fs::path private_key_file;
+            const Proxy sam_proxy;
+            const std::string_view accept_thread_name;
+        };
+
+        /**
+         * I2P options. If set then a thread will be started that will accept incoming I2P connections.
+         */
+        std::optional<I2P> i2p;
+    };
+
+    /**
+     * Start the necessary threads for sockets IO.
+     */
+    void StartSocketsThreads(const Options& options);
+
+    /**
+     * Join (wait for) the threads started by `StartSocketsThreads()` to exit.
+     */
+    void JoinSocketsThreads();
+
+    /**
+     * A more readable std::tuple<std::string, uint16_t> for host and port.
+     */
+    struct StringHostIntPort {
+        const std::string& host;
+        uint16_t port;
+    };
+
+    /**
+     * Make an outbound connection, save the socket internally and return a newly generated connection id.
+     * @param[in] to The address to connect to, either as CService or a host as string and port as
+     * an integer, if the later is used, then `proxy` must be valid.
+     * @param[in] is_important If true, then log failures with higher severity.
+     * @param[in] proxy Proxy to connect through, if set.
+     * @param[out] proxy_failed If `proxy` is valid and the connection failed because of the
+     * proxy, then it will be set to true.
+     * @param[out] me If the connection was successful then this is set to the address on the
+     * local side of the socket.
+     * @return Newly generated id, or std::nullopt if the operation fails.
+     */
+    std::optional<SockMan::Id> ConnectAndMakeId(const std::variant<CService, StringHostIntPort>& to,
+                                                bool is_important,
+                                                std::optional<Proxy> proxy,
+                                                bool& proxy_failed,
+                                                CService& me)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex, !m_unused_i2p_sessions_mutex);
+
+    /**
+     * Destroy a given connection by closing its socket and release resources occupied by it.
+     * @param[in] id Connection to destroy.
+     * @return Whether the connection existed and its socket was closed by this call.
+     */
+    bool CloseConnection(Id id)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Try to send some data over the given connection.
+     * @param[in] id Identifier of the connection.
+     * @param[in] data The data to send, it might happen that only a prefix of this is sent.
+     * @param[in] will_send_more Used as an optimization if the caller knows that they will
+     * be sending more data soon after this call.
+     * @param[out] errmsg If <0 is returned then this will contain a human readable message
+     * explaining the error.
+     * @retval >=0 The number of bytes actually sent.
+     * @retval <0 A permanent error has occurred.
+     */
+    ssize_t SendBytes(Id id,
+                      std::span<const unsigned char> data,
+                      bool will_send_more,
+                      std::string& errmsg) const
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Stop listening by closing all listening sockets.
+     */
+    void StopListening();
+
+    /**
+     * This is signaled when network activity should cease.
+     * A pointer to it is saved in `m_i2p_sam_session`, so make sure that
+     * the lifetime of `interruptNet` is not shorter than
+     * the lifetime of `m_i2p_sam_session`.
+     */
+    CThreadInterrupt interruptNet;
+
+protected:
+
+    /**
+     * During some tests mocked sockets are created outside of `SockMan`, make it
+     * possible to add those so that send/recv can be exercised.
+     * @param[in] id Connection id to add.
+     * @param[in,out] sock Socket to associate with the added connection.
+     */
+    void TestOnlyAddExistentConnection(Id id, std::unique_ptr<Sock>&& sock)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+private:
+
+    /**
+     * Cap on the size of `m_unused_i2p_sessions`, to ensure it does not
+     * unexpectedly use too much memory.
+     */
+    static constexpr size_t MAX_UNUSED_I2P_SESSIONS_SIZE{10};
+
+    //
+    // Pure virtual functions must be implemented by children classes.
+    //
+
+    /**
+     * Be notified when a new connection has been accepted.
+     * @param[in] id Id of the newly accepted connection.
+     * @param[in] me The address and port at our side of the connection.
+     * @param[in] them The address and port at the peer's side of the connection.
+     * @retval true The new connection was accepted at the higher level.
+     * @retval false The connection was refused at the higher level, so the
+     * associated socket and id should be discarded by `SockMan`.
+     */
+    virtual bool EventNewConnectionAccepted(Id id,
+                                            const CService& me,
+                                            const CService& them) = 0;
+
+    /**
+     * Called when the socket is ready to send data and `ShouldTryToSend()` has
+     * returned true. This is where the higher level code serializes its messages
+     * and calls `SockMan::SendBytes()`.
+     * @param[in] id Id of the connection whose socket is ready to send.
+     * @param[out] cancel_recv Should always be set upon return and if it is true,
+     * then the next attempt to receive data from that connection will be omitted.
+     */
+    virtual void EventReadyToSend(Id id, bool& cancel_recv) = 0;
+
+    /**
+     * Called when new data has been received.
+     * @param[in] id Connection for which the data arrived.
+     * @param[in] data Received data.
+     */
+    virtual void EventGotData(Id id, std::span<const uint8_t> data) = 0;
+
+    /**
+     * Called when the remote peer has sent an EOF on the socket. This is a graceful
+     * close of their writing side, we can still send and they will receive, if it
+     * makes sense at the application level.
+     * @param[in] id Connection whose socket got EOF.
+     */
+    virtual void EventGotEOF(Id id) = 0;
+
+    /**
+     * Called when we get an irrecoverable error trying to read from a socket.
+     * @param[in] id Connection whose socket got an error.
+     * @param[in] errmsg Message describing the error.
+     */
+    virtual void EventGotPermanentReadError(Id id, const std::string& errmsg) = 0;
+
+    //
+    // Non-pure virtual functions can be overridden by children classes or left
+    // alone to use the default implementation from SockMan.
+    //
+
+    /**
+     * Can be used to temporarily pause sends on a connection.
+     * SockMan would only call EventReadyToSend() if this returns true.
+     * The implementation in SockMan always returns true.
+     * @param[in] id Connection for which to confirm or omit the next call to EventReadyToSend().
+     */
+    virtual bool ShouldTryToSend(Id id) const;
+
+    /**
+     * SockMan would only call Recv() on a connection's socket if this returns true.
+     * Can be used to temporarily pause receives on a connection.
+     * The implementation in SockMan always returns true.
+     * @param[in] id Connection for which to confirm or omit the next receive.
+     */
+    virtual bool ShouldTryToRecv(Id id) const;
+
+    /**
+     * SockMan has completed the current send+recv iteration for a given connection.
+     * It will do another send+recv for this connection after processing all other connections.
+     * Can be used to execute periodic tasks for a given connection.
+     * The implementation in SockMan does nothing.
+     * @param[in] id Connection for which send+recv has been done.
+     */
+    virtual void EventIOLoopCompletedForOne(Id id);
+
+    /**
+     * SockMan has completed send+recv for all connections.
+     * Can be used to execute periodic tasks for all connections, like closing
+     * connections due to higher level logic.
+     * The implementation in SockMan does nothing.
+     */
+    virtual void EventIOLoopCompletedForAll();
+
+    /**
+     * Be notified of a change in the state of the I2P connectivity.
+     * The default behavior, implemented by `SockMan`, is to ignore this event.
+     * @param[in] addr The address we started or stopped listening on.
+     * @param[in] new_status New status.
+     */
+    virtual void EventI2PStatus(const CService& addr, I2PStatus new_status);
+
+    /**
+     * The sockets used by a connection - a data socket and an optional I2P session socket.
+     */
+    struct ConnectionSockets {
+        explicit ConnectionSockets(std::unique_ptr<Sock>&& s)
+            : sock{std::move(s)}
+        {
+        }
+
+        explicit ConnectionSockets(std::shared_ptr<Sock>&& s, std::unique_ptr<i2p::sam::Session>&& sess)
+            : sock{std::move(s)},
+              i2p_transient_session{std::move(sess)}
+        {
+        }
+
+        /**
+         * Mutex that serializes the Send() and Recv() calls on `sock`.
+         */
+        Mutex mutex;
+
+        /**
+         * Underlying socket.
+         * `shared_ptr` (instead of `unique_ptr`) is used to avoid premature close of the
+         * underlying file descriptor by one thread while another thread is poll(2)-ing
+         * it for activity.
+         * @see https://github.com/bitcoin/bitcoin/issues/21744 for details.
+         */
+        std::shared_ptr<Sock> sock;
+
+        /**
+         * When transient I2P sessions are used, then each connection has its own session, otherwise
+         * all connections use the session from `m_i2p_sam_session` and share the same I2P address.
+         * I2P sessions involve a data/transport socket (in `sock`) and a control socket
+         * (in `i2p_transient_session`). For transient sessions, once the data socket `sock` is
+         * closed, the control socket is not going to be used anymore and would be just taking
+         * resources. Storing it here makes its deletion together with `sock` automatic.
+         */
+        std::unique_ptr<i2p::sam::Session> i2p_transient_session;
+    };
+
+    /**
+     * Info about which socket has which event ready and its connection id.
+     */
+    struct IOReadiness {
+        /**
+         * Map of socket -> socket events. For example:
+         * socket1 -> { requested = SEND|RECV, occurred = RECV }
+         * socket2 -> { requested = SEND, occurred = SEND }
+         */
+        Sock::EventsPerSock events_per_sock;
+
+        /**
+         * Map of socket -> connection id (in `m_connected`). For example
+         * socket1 -> id=23
+         * socket2 -> id=56
+         */
+        std::unordered_map<Sock::EventsPerSock::key_type,
+                           SockMan::Id,
+                           Sock::HashSharedPtrSock,
+                           Sock::EqualSharedPtrSock>
+            ids_per_sock;
+    };
+
+    /**
+     * Accept incoming I2P connections in a loop and call
+     * `EventNewConnectionAccepted()` for each new connection.
+     */
+    void ThreadI2PAccept()
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Check connected and listening sockets for IO readiness and process them accordingly.
+     */
+    void ThreadSocketHandler()
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Accept a connection.
+     * @param[in] listen_sock Socket on which to accept the connection.
+     * @param[out] addr Address of the peer that was accepted.
+     * @return Newly created socket for the accepted connection.
+     */
+    std::unique_ptr<Sock> AcceptConnection(const Sock& listen_sock, CService& addr);
+
+    /**
+     * After a new socket with a peer has been created, configure its flags,
+     * make a new connection id and call `EventNewConnectionAccepted()`.
+     * @param[in] sock The newly created socket.
+     * @param[in] me Address at our end of the connection.
+     * @param[in] them Address of the new peer.
+     */
+    void NewSockAccepted(std::unique_ptr<Sock>&& sock, const CService& me, const CService& them)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Generate an id for a newly created connection.
+     */
+    Id GetNewId();
+
+    /**
+     * Generate a collection of sockets to check for IO readiness.
+     * @return Sockets to check for readiness plus an aux map to find the
+     * corresponding connection id given a socket.
+     */
+    IOReadiness GenerateWaitSockets()
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Do the read/write for connected sockets that are ready for IO.
+     * @param[in] io_readiness Which sockets are ready and their connection ids.
+     */
+    void SocketHandlerConnected(const IOReadiness& io_readiness)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Accept incoming connections, one from each read-ready listening socket.
+     * @param[in] events_per_sock Sockets that are ready for IO.
+     */
+    void SocketHandlerListening(const Sock::EventsPerSock& events_per_sock)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * Retrieve an entry from m_connected.
+     * @param[in] id Connection id to search for.
+     * @return ConnectionSockets for the given connection id or empty shared_ptr if not found.
+     */
+    std::shared_ptr<ConnectionSockets> GetConnectionSockets(Id id) const
+        EXCLUSIVE_LOCKS_REQUIRED(!m_connected_mutex);
+
+    /**
+     * The id to assign to the next created connection. Used to generate ids of connections.
+     */
+    std::atomic<Id> m_next_id{0};
+
+    /**
+     * Thread that sends to and receives from sockets and accepts connections.
+     */
+    std::thread m_thread_socket_handler;
+
+    /**
+     * Thread that accepts incoming I2P connections in a loop, can be stopped via `interruptNet`.
+     */
+    std::thread m_thread_i2p_accept;
+
+    /**
+     * Mutex protecting m_i2p_sam_sessions.
+     */
+    Mutex m_unused_i2p_sessions_mutex;
+
+    /**
+     * A pool of created I2P SAM transient sessions that should be used instead
+     * of creating new ones in order to reduce the load on the I2P network.
+     * Creating a session in I2P is not cheap, thus if this is not empty, then
+     * pick an entry from it instead of creating a new session. If connecting to
+     * a host fails, then the created session is put to this pool for reuse.
+     */
+    std::queue<std::unique_ptr<i2p::sam::Session>> m_unused_i2p_sessions GUARDED_BY(m_unused_i2p_sessions_mutex);
+
+    /**
+     * I2P SAM session.
+     * Used to accept incoming and make outgoing I2P connections from a persistent
+     * address.
+     */
+    std::unique_ptr<i2p::sam::Session> m_i2p_sam_session;
+
+    /**
+     * List of listening sockets.
+     */
+    std::vector<std::shared_ptr<Sock>> m_listen;
+
+    mutable Mutex m_connected_mutex;
+
+    /**
+     * Sockets for existent connections.
+     * The `shared_ptr` makes it possible to create a snapshot of this by simply copying
+     * it (under `m_connected_mutex`).
+     */
+    std::unordered_map<Id, std::shared_ptr<ConnectionSockets>> m_connected GUARDED_BY(m_connected_mutex);
+};
+
+#endif // BITCOIN_COMMON_SOCKMAN_H

src/httprpc.cpp

@@ -9,8 +9,10 @@
 #include <httpserver.h>
 #include <logging.h>
 #include <netaddress.h>
+#include <node/context.h>
 #include <rpc/protocol.h>
 #include <rpc/server.h>
+#include <scheduler.h>
 #include <util/fs.h>
 #include <util/fs_helpers.h>
 #include <util/strencodings.h>
@@ -26,6 +28,8 @@
 #include <string>
 #include <vector>
 
+using node::NodeContext;
+using http_bitcoin::HTTPRequest;
 using util::SplitString;
 using util::TrimStringView;
 
@@ -38,22 +42,16 @@ static const char* WWW_AUTH_HEADER_DATA = "Basic realm=\"jsonrpc\"";
 class HTTPRPCTimer : public RPCTimerBase
 {
 public:
-    HTTPRPCTimer(struct event_base* eventBase, std::function<void()>& func, int64_t millis) :
-        ev(eventBase, false, func)
+    HTTPRPCTimer(NodeContext* context, std::function<void()>& func, int64_t millis)
     {
-        struct timeval tv;
-        tv.tv_sec = millis/1000;
-        tv.tv_usec = (millis%1000)*1000;
-        ev.trigger(&tv);
+        context->scheduler->scheduleFromNow(func, std::chrono::milliseconds(millis));
     }
-private:
-    HTTPEvent ev;
 };
 
 class HTTPRPCTimerInterface : public RPCTimerInterface
 {
 public:
-    explicit HTTPRPCTimerInterface(struct event_base* _base) : base(_base)
+    explicit HTTPRPCTimerInterface(const std::any& context) : m_context(std::any_cast<NodeContext*>(context))
     {
     }
     const char* Name() override
@@ -62,10 +60,10 @@ public:
     }
     RPCTimerBase* NewTimer(std::function<void()>& func, int64_t millis) override
     {
-        return new HTTPRPCTimer(base, func, millis);
+        return new HTTPRPCTimer(m_context, func, millis);
     }
 private:
-    struct event_base* base;
+    NodeContext* m_context;
 };
 
 
@@ -85,7 +83,7 @@ static void JSONErrorReply(HTTPRequest* req, UniValue objError, const JSONRPCReq
     Assume(jreq.m_json_version != JSONRPCVersion::V2);
 
     // Send error reply from json-rpc error object
-    int nStatus = HTTP_INTERNAL_SERVER_ERROR;
+    HTTPStatusCode nStatus = HTTP_INTERNAL_SERVER_ERROR;
     int code = objError.find_value("code").getInt<int>();
 
     if (code == RPC_INVALID_REQUEST)
@@ -156,7 +154,7 @@ static bool RPCAuthorized(const std::string& strAuth, std::string& strAuthUserna
 static bool HTTPReq_JSONRPC(const std::any& context, HTTPRequest* req)
 {
     // JSONRPC handles only POST
-    if (req->GetRequestMethod() != HTTPRequest::POST) {
+    if (req->GetRequestMethod() != HTTPRequestMethod::POST) {
         req->WriteReply(HTTP_BAD_METHOD, "JSONRPC server handles only POST requests");
         return false;
     }
@@ -370,9 +368,7 @@ bool StartHTTPRPC(const std::any& context)
     if (g_wallet_init_interface.HasWalletSupport()) {
         RegisterHTTPHandler("/wallet/", false, handle_rpc);
     }
-    struct event_base* eventBase = EventBase();
-    assert(eventBase);
-    httpRPCTimerInterface = std::make_unique<HTTPRPCTimerInterface>(eventBase);
+    httpRPCTimerInterface = std::make_unique<HTTPRPCTimerInterface>(context);
     RPCSetTimerInterface(httpRPCTimerInterface.get());
     return true;
 }

src/httpserver.h

@@ -6,10 +6,17 @@
 #define BITCOIN_HTTPSERVER_H
 
 #include <functional>
+#include <map>
 #include <optional>
 #include <span>
 #include <string>
 
+#include <rpc/protocol.h>
+#include <common/sockman.h>
+#include <util/strencodings.h>
+#include <util/string.h>
+#include <util/time.h>
+
 namespace util {
 class SignalInterrupt;
 } // namespace util
@@ -27,10 +34,272 @@ static const int DEFAULT_HTTP_WORKQUEUE=64;
 
 static const int DEFAULT_HTTP_SERVER_TIMEOUT=30;
 
-struct evhttp_request;
-struct event_base;
-class CService;
-class HTTPRequest;
+enum HTTPRequestMethod {
+    UNKNOWN,
+    GET,
+    POST,
+    HEAD,
+    PUT
+};
+
+/** Event handler closure.
+ */
+class HTTPClosure
+{
+public:
+    virtual void operator()() = 0;
+    virtual ~HTTPClosure() = default;
+};
+
+namespace http_bitcoin {
+using util::LineReader;
+using NodeId = SockMan::Id;
+
+// shortest valid request line, used by libevent in evhttp_parse_request_line()
+static const size_t MIN_REQUEST_LINE_LENGTH{strlen("GET / HTTP/1.0")};
+// maximum size of http request (request line + headers)
+// see https://github.com/bitcoin/bitcoin/issues/6425
+static const size_t MAX_HEADERS_SIZE{8192};
+
+class HTTPHeaders
+{
+public:
+    std::optional<std::string> Find(const std::string key) const;
+    void Write(const std::string key, const std::string value);
+    void Remove(const std::string key);
+    bool Read(util::LineReader& reader);
+    std::string Stringify() const;
+
+private:
+    std::map<std::string, std::string, util::CaseInsensitiveComparator> m_map;
+};
+
+class HTTPResponse
+{
+public:
+    int m_version_major;
+    int m_version_minor;
+    HTTPStatusCode m_status;
+    std::string m_reason;
+    HTTPHeaders m_headers;
+    std::vector<std::byte> m_body;
+    bool m_keep_alive{false};
+
+    std::string StringifyHeaders() const;
+};
+
+class HTTPClient;
+
+class HTTPRequest
+{
+public:
+    std::string m_method;
+    std::string m_target;
+    // Default protocol version is used by error responses to unreadable requests
+    int m_version_major{1};
+    int m_version_minor{1};
+    HTTPHeaders m_headers;
+    std::string m_body;
+
+    // Keep a pointer to the client that made the request so
+    // we know who to respond to.
+    std::shared_ptr<HTTPClient> m_client;
+    explicit HTTPRequest(std::shared_ptr<HTTPClient> client) : m_client(client) {};
+    // Null client for unit tests
+    explicit HTTPRequest() : m_client(nullptr) {};
+
+    // Readers return false if they need more data from the
+    // socket to parse properly. They throw errors if
+    // the data is invalid.
+    bool LoadControlData(LineReader& reader);
+    bool LoadHeaders(LineReader& reader);
+    bool LoadBody(LineReader& reader);
+
+    // These methods reimplement the API from http_libevent::HTTPRequest
+    // for downstream JSONRPC and REST modules.
+    std::string GetURI() const {return m_target;};
+    CService GetPeer() const;
+    HTTPRequestMethod GetRequestMethod() const;
+    std::optional<std::string> GetQueryParameter(const std::string& key) const;
+    std::pair<bool, std::string> GetHeader(const std::string& hdr) const;
+    std::string ReadBody() const {return m_body;};
+    void WriteHeader(const std::string& hdr, const std::string& value);
+
+    // Response headers may be set in advance before response body is known
+    HTTPHeaders m_response_headers;
+    void WriteReply(HTTPStatusCode status, std::span<const std::byte> reply_body = {});
+    void WriteReply(HTTPStatusCode status, const char* reply_body)
+    {
+        auto reply_body_view = std::string_view(reply_body);
+        std::span<const std::byte> byte_span(reinterpret_cast<const std::byte*>(reply_body_view.data()), reply_body_view.size());
+        WriteReply(status, byte_span);
+    }
+    void WriteReply(HTTPStatusCode status, const std::string& reply_body)
+    {
+        std::span<const std::byte> byte_span{reinterpret_cast<const std::byte*>(reply_body.data()), reply_body.size()};
+        WriteReply(status, byte_span);
+    }
+};
+
+std::optional<std::string> GetQueryParameterFromUri(const std::string& uri, const std::string& key);
+
+class HTTPServer;
+
+class HTTPClient
+{
+public:
+    // ID provided by SockMan, inherited by HTTPServer
+    NodeId m_node_id;
+    // Remote address of connected client
+    CService m_addr;
+    // IP:port of connected client, cached for logging purposes
+    std::string m_origin;
+    // Pointer back to the server so we can call Sockman I/O methods from the client
+    // Ok to remain null for unit tests.
+    HTTPServer* m_server;
+
+    // In lieu of an intermediate transport class like p2p uses,
+    // we copy data from the socket buffer to the client object
+    // and attempt to read HTTP requests from here.
+    std::vector<std::byte> m_recv_buffer{};
+
+    // Response data destined for this client.
+    // Written to directly by http worker threads, read and erased by Sockman I/O
+    Mutex m_send_mutex;
+    std::vector<std::byte> m_send_buffer GUARDED_BY(m_send_mutex);
+    // Set true by worker threads after writing a response to m_send_buffer.
+    // Set false by the Sockman I/O thread after flushing m_send_buffer.
+    // Checked in the Sockman I/O loop to avoid locking m_send_mutex if there's nothing to send.
+    std::atomic_bool m_send_ready{false};
+
+    // Set to true when we receive request data and set to false once m_send_buffer is cleared.
+    // Checked during DisconnectClients(). All of these operations take place in the Sockman I/O loop.
+    bool m_prevent_disconnect{false};
+
+    // Client request to keep connection open after all requests have been responded to.
+    // Set by (potentially multiple) worker threads and checked in the Sockman I/O loop.
+    std::atomic_bool m_keep_alive{false};
+
+    // Flag this client for disconnection on next loop
+    bool m_disconnect{false};
+
+    // Timestamp of last receive activity, used for -rpcservertimeout
+    SteadySeconds m_idle_since;
+
+    explicit HTTPClient(NodeId node_id, CService addr) : m_node_id(node_id), m_addr(addr)
+    {
+        m_origin = addr.ToStringAddrPort();
+    };
+
+    // Try to read an HTTP request from the receive buffer
+    bool ReadRequest(std::unique_ptr<HTTPRequest>& req);
+
+    // Push data from m_send_buffer to the connected socket via m_server
+    // Returns false if we are done with this client and Sockman can
+    // therefore skip the next read operation from it.
+    bool SendBytesFromBuffer() EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex);
+
+    // Disable copies (should only be used as shared pointers)
+    HTTPClient(const HTTPClient&) = delete;
+    HTTPClient& operator=(const HTTPClient&) = delete;
+};
+
+class HTTPServer : public SockMan
+{
+private:
+    void CloseConnectionInternal(std::shared_ptr<HTTPClient>& client);
+
+public:
+    explicit HTTPServer(std::function<void(std::unique_ptr<HTTPRequest>)> func) : m_request_dispatcher(func) {};
+
+    // Set in the Sockman I/O loop and only checked by main thread when shutting
+    // down to wait for all clients to be disconnected.
+    std::atomic_bool m_no_clients{true};
+    //! Connected clients with live HTTP connections
+    std::unordered_map<NodeId, std::shared_ptr<HTTPClient>> m_connected_clients;
+
+    // What to do with HTTP requests once received, validated and parsed
+    std::function<void(std::unique_ptr<HTTPRequest>)> m_request_dispatcher;
+
+    std::shared_ptr<HTTPClient> GetClientById(NodeId node_id) const;
+
+    // Close underlying connections where flagged
+    void DisconnectClients();
+
+    // Flag used during shutdown to bypass keep-alive flag.
+    // Set by main thread and read by Sockman I/O thread
+    std::atomic_bool m_disconnect_all_clients{false};
+
+    // Idle timeout after which clients are disconnected
+    std::chrono::seconds m_rpcservertimeout;
+
+    /**
+     * Be notified when a new connection has been accepted.
+     * @param[in] node_id Id of the newly accepted connection.
+     * @param[in] me The address and port at our side of the connection.
+     * @param[in] them The address and port at the peer's side of the connection.
+     * @retval true The new connection was accepted at the higher level.
+     * @retval false The connection was refused at the higher level, so the
+     * associated socket and node_id should be discarded by `SockMan`.
+     */
+    virtual bool EventNewConnectionAccepted(NodeId node_id, const CService& me, const CService& them) override;
+
+    /**
+     * Called when the socket is ready to send data and `ShouldTryToSend()` has
+     * returned true. This is where the higher level code serializes its messages
+     * and calls `SockMan::SendBytes()`.
+     * @param[in] node_id Id of the node whose socket is ready to send.
+     * @param[out] cancel_recv Should always be set upon return and if it is true,
+     * then the next attempt to receive data from that node will be omitted.
+     */
+    virtual void EventReadyToSend(NodeId node_id, bool& cancel_recv) override;
+
+    /**
+     * Called when new data has been received.
+     * @param[in] node_id Connection for which the data arrived.
+     * @param[in] data Received data.
+     */
+    virtual void EventGotData(NodeId node_id, std::span<const uint8_t> data) override;
+
+    /**
+     * Called when the remote peer has sent an EOF on the socket. This is a graceful
+     * close of their writing side, we can still send and they will receive, if it
+     * makes sense at the application level.
+     * @param[in] node_id Node whose socket got EOF.
+     */
+    virtual void EventGotEOF(NodeId node_id) override;
+
+    /**
+     * Called when we get an irrecoverable error trying to read from a socket.
+     * @param[in] node_id Node whose socket got an error.
+     * @param[in] errmsg Message describing the error.
+     */
+    virtual void EventGotPermanentReadError(NodeId node_id, const std::string& errmsg) override;
+
+    /**
+     * SockMan has completed send+recv for all nodes.
+     * Can be used to execute periodic tasks for all nodes, like disconnecting
+     * nodes due to higher level logic.
+     * The implementation in SockMan does nothing.
+     */
+    virtual void EventIOLoopCompletedForAll() override;
+
+    /**
+     * Can be used to temporarily pause sends on a connection.
+     * SockMan would only call EventReadyToSend() if this returns true.
+     * The implementation in SockMan always returns true.
+     * @param[in] node_id Connection for which to confirm or omit the next call to EventReadyToSend().
+     */
+    virtual bool ShouldTryToSend(NodeId node_id) const override;
+
+    /**
+     * SockMan would only call Recv() on a connection's socket if this returns true.
+     * Can be used to temporarily pause receives on a connection.
+     * The implementation in SockMan always returns true.
+     * @param[in] node_id Connection for which to confirm or omit the next receive.
+     */
+    virtual bool ShouldTryToRecv(NodeId node_id) const override;
+};
 
 /** Initialize HTTP server.
  * Call this before RegisterHTTPHandler or EventBase().
@@ -45,12 +314,10 @@ void StartHTTPServer();
 void InterruptHTTPServer();
 /** Stop HTTP server */
 void StopHTTPServer();
-
-/** Change logging level for libevent. */
-void UpdateHTTPServerLogging(bool enable);
+} // namespace http_bitcoin
 
 /** Handler for requests to a certain HTTP path */
-typedef std::function<bool(HTTPRequest* req, const std::string &)> HTTPRequestHandler;
+typedef std::function<bool(http_bitcoin::HTTPRequest* req, const std::string&)> HTTPRequestHandler;
 /** Register handler for prefix.
  * If multiple handlers match a prefix, the first-registered one will
  * be invoked.
@@ -59,136 +326,4 @@ void RegisterHTTPHandler(const std::string &prefix, bool exactMatch, const HTTPR
 /** Unregister handler for prefix */
 void UnregisterHTTPHandler(const std::string &prefix, bool exactMatch);
 
-/** Return evhttp event base. This can be used by submodules to
- * queue timers or custom events.
- */
-struct event_base* EventBase();
-
-/** In-flight HTTP request.
- * Thin C++ wrapper around evhttp_request.
- */
-class HTTPRequest
-{
-private:
-    struct evhttp_request* req;
-    const util::SignalInterrupt& m_interrupt;
-    bool replySent;
-
-public:
-    explicit HTTPRequest(struct evhttp_request* req, const util::SignalInterrupt& interrupt, bool replySent = false);
-    ~HTTPRequest();
-
-    enum RequestMethod {
-        UNKNOWN,
-        GET,
-        POST,
-        HEAD,
-        PUT
-    };
-
-    /** Get requested URI.
-     */
-    std::string GetURI() const;
-
-    /** Get CService (address:ip) for the origin of the http request.
-     */
-    CService GetPeer() const;
-
-    /** Get request method.
-     */
-    RequestMethod GetRequestMethod() const;
-
-    /** Get the query parameter value from request uri for a specified key, or std::nullopt if the
-     * key is not found.
-     *
-     * If the query string contains duplicate keys, the first value is returned. Many web frameworks
-     * would instead parse this as an array of values, but this is not (yet) implemented as it is
-     * currently not needed in any of the endpoints.
-     *
-     * @param[in] key represents the query parameter of which the value is returned
-     */
-    std::optional<std::string> GetQueryParameter(const std::string& key) const;
-
-    /**
-     * Get the request header specified by hdr, or an empty string.
-     * Return a pair (isPresent,string).
-     */
-    std::pair<bool, std::string> GetHeader(const std::string& hdr) const;
-
-    /**
-     * Read request body.
-     *
-     * @note As this consumes the underlying buffer, call this only once.
-     * Repeated calls will return an empty string.
-     */
-    std::string ReadBody();
-
-    /**
-     * Write output header.
-     *
-     * @note call this before calling WriteErrorReply or Reply.
-     */
-    void WriteHeader(const std::string& hdr, const std::string& value);
-
-    /**
-     * Write HTTP reply.
-     * nStatus is the HTTP status code to send.
-     * reply is the body of the reply. Keep it empty to send a standard message.
-     *
-     * @note Can be called only once. As this will give the request back to the
-     * main thread, do not call any other HTTPRequest methods after calling this.
-     */
-    void WriteReply(int nStatus, std::string_view reply = "")
-    {
-        WriteReply(nStatus, std::as_bytes(std::span{reply}));
-    }
-    void WriteReply(int nStatus, std::span<const std::byte> reply);
-};
-
-/** Get the query parameter value from request uri for a specified key, or std::nullopt if the key
- * is not found.
- *
- * If the query string contains duplicate keys, the first value is returned. Many web frameworks
- * would instead parse this as an array of values, but this is not (yet) implemented as it is
- * currently not needed in any of the endpoints.
- *
- * Helper function for HTTPRequest::GetQueryParameter.
- *
- * @param[in] uri is the entire request uri
- * @param[in] key represents the query parameter of which the value is returned
- */
-std::optional<std::string> GetQueryParameterFromUri(const char* uri, const std::string& key);
-
-/** Event handler closure.
- */
-class HTTPClosure
-{
-public:
-    virtual void operator()() = 0;
-    virtual ~HTTPClosure() = default;
-};
-
-/** Event class. This can be used either as a cross-thread trigger or as a timer.
- */
-class HTTPEvent
-{
-public:
-    /** Create a new event.
-     * deleteWhenTriggered deletes this event object after the event is triggered (and the handler called)
-     * handler is the handler to call when the event is triggered.
-     */
-    HTTPEvent(struct event_base* base, bool deleteWhenTriggered, const std::function<void()>& handler);
-    ~HTTPEvent();
-
-    /** Trigger the event. If tv is 0, trigger it immediately. Otherwise trigger it after
-     * the given time has elapsed.
-     */
-    void trigger(struct timeval* tv);
-
-    bool deleteWhenTriggered;
-    std::function<void()> handler;
-private:
-    struct event* ev;
-};
-
 #endif // BITCOIN_HTTPSERVER_H

src/init.cpp

@@ -121,6 +121,10 @@ using common::AmountErrMsg;
 using common::InvalidPortErrMsg;
 using common::ResolveErrMsg;
 
+using http_bitcoin::InitHTTPServer;
+using http_bitcoin::InterruptHTTPServer;
+using http_bitcoin::StartHTTPServer;
+using http_bitcoin::StopHTTPServer;
 using node::ApplyArgsManOptions;
 using node::BlockManager;
 using node::CalculateCacheSizes;

src/net.h

@@ -9,6 +9,7 @@
 #include <bip324.h>
 #include <chainparams.h>
 #include <common/bloom.h>
+#include <common/sockman.h>
 #include <compat/compat.h>
 #include <consensus/amount.h>
 #include <crypto/siphash.h>
@@ -41,8 +42,8 @@
 #include <map>
 #include <memory>
 #include <optional>
-#include <queue>
 #include <thread>
+#include <unordered_map>
 #include <unordered_set>
 #include <vector>
 
@@ -94,7 +95,7 @@ static const size_t DEFAULT_MAXSENDBUFFER    = 1 * 1000;
 
 static constexpr bool DEFAULT_V2_TRANSPORT{true};
 
-typedef int64_t NodeId;
+using NodeId = SockMan::Id;
 
 struct AddedNodeParams {
     std::string m_added_node;
@@ -662,7 +663,6 @@ public:
 struct CNodeOptions
 {
     NetPermissionFlags permission_flags = NetPermissionFlags::None;
-    std::unique_ptr<i2p::sam::Session> i2p_sam_session = nullptr;
     bool prefer_evict = false;
     size_t recv_flood_size{DEFAULT_MAXRECEIVEBUFFER * 1000};
     bool use_v2transport = false;
@@ -678,16 +678,6 @@ public:
 
     const NetPermissionFlags m_permission_flags;
 
-    /**
-     * Socket used for communication with the node.
-     * May not own a Sock object (after `CloseSocketDisconnect()` or during tests).
-     * `shared_ptr` (instead of `unique_ptr`) is used to avoid premature close of
-     * the underlying file descriptor by one thread while another thread is
-     * poll(2)-ing it for activity.
-     * @see https://github.com/bitcoin/bitcoin/issues/21744 for details.
-     */
-    std::shared_ptr<Sock> m_sock GUARDED_BY(m_sock_mutex);
-
     /** Sum of GetMemoryUsage of all vSendMsg entries. */
     size_t m_send_memusage GUARDED_BY(cs_vSend){0};
     /** Total number of bytes sent on the wire to this peer. */
@@ -695,7 +685,6 @@ public:
     /** Messages still to be fed to m_transport->SetMessageToSend. */
     std::deque<CSerializedNetMsg> vSendMsg GUARDED_BY(cs_vSend);
     Mutex cs_vSend;
-    Mutex m_sock_mutex;
     Mutex cs_vRecv;
 
     uint64_t nRecvBytes GUARDED_BY(cs_vRecv){0};
@@ -879,7 +868,6 @@ public:
     std::atomic<std::chrono::microseconds> m_min_ping_time{std::chrono::microseconds::max()};
 
     CNode(NodeId id,
-          std::shared_ptr<Sock> sock,
           const CAddress& addrIn,
           uint64_t nKeyedNetGroupIn,
           uint64_t nLocalHostNonceIn,
@@ -941,8 +929,6 @@ public:
         nRefCount--;
     }
 
-    void CloseSocketDisconnect() EXCLUSIVE_LOCKS_REQUIRED(!m_sock_mutex);
-
     void CopyStats(CNodeStats& stats) EXCLUSIVE_LOCKS_REQUIRED(!m_subver_mutex, !m_addr_local_mutex, !cs_vSend, !cs_vRecv);
 
     std::string ConnectionTypeAsString() const { return ::ConnectionTypeAsString(m_conn_type); }
@@ -987,18 +973,6 @@ private:
 
     mapMsgTypeSize mapSendBytesPerMsgType GUARDED_BY(cs_vSend);
     mapMsgTypeSize mapRecvBytesPerMsgType GUARDED_BY(cs_vRecv);
-
-    /**
-     * If an I2P session is created per connection (for outbound transient I2P
-     * connections) then it is stored here so that it can be destroyed when the
-     * socket is closed. I2P sessions involve a data/transport socket (in `m_sock`)
-     * and a control socket (in `m_i2p_sam_session`). For transient sessions, once
-     * the data socket is closed, the control socket is not going to be used anymore
-     * and is just taking up resources. So better close it as soon as `m_sock` is
-     * closed.
-     * Otherwise this unique_ptr is empty.
-     */
-    std::unique_ptr<i2p::sam::Session> m_i2p_sam_session GUARDED_BY(m_sock_mutex);
 };
 
 /**
@@ -1048,7 +1022,7 @@ protected:
     ~NetEventsInterface() = default;
 };
 
-class CConnman
+class CConnman : private SockMan
 {
 public:
 
@@ -1136,7 +1110,7 @@ public:
     bool GetNetworkActive() const { return fNetworkActive; };
     bool GetUseAddrmanOutgoing() const { return m_use_addrman_outgoing; };
     void SetNetworkActive(bool active);
-    void OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant&& grant_outbound, const char* strDest, ConnectionType conn_type, bool use_v2transport) EXCLUSIVE_LOCKS_REQUIRED(!m_unused_i2p_sessions_mutex);
+    void OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant&& grant_outbound, const char* strDest, ConnectionType conn_type, bool use_v2transport);
     bool CheckIncomingNonce(uint64_t nonce);
     void ASMapHealthCheck();
 
@@ -1151,7 +1125,7 @@ public:
     void ForEachNode(const NodeFn& func)
     {
         LOCK(m_nodes_mutex);
-        for (auto&& node : m_nodes) {
+        for (auto& [_, node] : m_nodes) {
             if (NodeFullyConnected(node))
                 func(node);
         }
@@ -1160,7 +1134,7 @@ public:
     void ForEachNode(const NodeFn& func) const
     {
         LOCK(m_nodes_mutex);
-        for (auto&& node : m_nodes) {
+        for (auto& [_, node] : m_nodes) {
             if (NodeFullyConnected(node))
                 func(node);
         }
@@ -1221,7 +1195,7 @@ public:
      *                          - Max total outbound connection capacity filled
      *                          - Max connection capacity for type is filled
      */
-    bool AddConnection(const std::string& address, ConnectionType conn_type, bool use_v2transport) EXCLUSIVE_LOCKS_REQUIRED(!m_unused_i2p_sessions_mutex);
+    bool AddConnection(const std::string& address, ConnectionType conn_type, bool use_v2transport);
 
     size_t GetNodeCount(ConnectionDirection) const;
     std::map<CNetAddr, LocalServiceInfo> getNetLocalAddresses() const;
@@ -1273,81 +1247,71 @@ public:
     bool MultipleManualOrFullOutboundConns(Network net) const EXCLUSIVE_LOCKS_REQUIRED(m_nodes_mutex);
 
 private:
-    struct ListenSocket {
-    public:
-        std::shared_ptr<Sock> sock;
-        inline void AddSocketPermissionFlags(NetPermissionFlags& flags) const { NetPermissions::AddFlag(flags, m_permissions); }
-        ListenSocket(std::shared_ptr<Sock> sock_, NetPermissionFlags permissions_)
-            : sock{sock_}, m_permissions{permissions_}
-        {
-        }
-
-    private:
-        NetPermissionFlags m_permissions;
-    };
-
     //! returns the time left in the current max outbound cycle
     //! in case of no limit, it will always return 0
     std::chrono::seconds GetMaxOutboundTimeLeftInCycle_() const EXCLUSIVE_LOCKS_REQUIRED(m_total_bytes_sent_mutex);
 
-    bool BindListenPort(const CService& bindAddr, bilingual_str& strError, NetPermissionFlags permissions);
     bool Bind(const CService& addr, unsigned int flags, NetPermissionFlags permissions);
     bool InitBinds(const Options& options);
 
-    void ThreadOpenAddedConnections() EXCLUSIVE_LOCKS_REQUIRED(!m_added_nodes_mutex, !m_unused_i2p_sessions_mutex, !m_reconnections_mutex);
+    void ThreadOpenAddedConnections() EXCLUSIVE_LOCKS_REQUIRED(!m_added_nodes_mutex, !m_reconnections_mutex);
     void AddAddrFetch(const std::string& strDest) EXCLUSIVE_LOCKS_REQUIRED(!m_addr_fetches_mutex);
-    void ProcessAddrFetch() EXCLUSIVE_LOCKS_REQUIRED(!m_addr_fetches_mutex, !m_unused_i2p_sessions_mutex);
-    void ThreadOpenConnections(std::vector<std::string> connect, Span<const std::string> seed_nodes) EXCLUSIVE_LOCKS_REQUIRED(!m_addr_fetches_mutex, !m_added_nodes_mutex, !m_nodes_mutex, !m_unused_i2p_sessions_mutex, !m_reconnections_mutex);
+    void ProcessAddrFetch() EXCLUSIVE_LOCKS_REQUIRED(!m_addr_fetches_mutex);
+    void ThreadOpenConnections(std::vector<std::string> connect, Span<const std::string> seed_nodes) EXCLUSIVE_LOCKS_REQUIRED(!m_addr_fetches_mutex, !m_added_nodes_mutex, !m_nodes_mutex, !m_reconnections_mutex);
     void ThreadMessageHandler() EXCLUSIVE_LOCKS_REQUIRED(!mutexMsgProc);
-    void ThreadI2PAcceptIncoming();
-    void AcceptConnection(const ListenSocket& hListenSocket);
+
+    /// Whether we are currently advertising our I2P address (via `AddLocal()`).
+    bool m_i2p_advertising_listen_addr{false};
+
+    virtual void EventI2PStatus(const CService& addr, SockMan::I2PStatus new_status) override;
 
     /**
-     * Create a `CNode` object from a socket that has just been accepted and add the node to
-     * the `m_nodes` member.
-     * @param[in] sock Connected socket to communicate with the peer.
-     * @param[in] permission_flags The peer's permissions.
-     * @param[in] addr_bind The address and port at our side of the connection.
-     * @param[in] addr The address and port at the peer's side of the connection.
+     * Create a `CNode` object and add it to the `m_nodes` member.
+     * @param[in] id Id of the newly accepted connection.
+     * @param[in] me The address and port at our side of the connection.
+     * @param[in] them The address and port at the peer's side of the connection.
+     * @retval true on success
+     * @retval false on failure, meaning that the associated socket and node_id should be discarded
      */
-    void CreateNodeFromAcceptedSocket(std::unique_ptr<Sock>&& sock,
-                                      NetPermissionFlags permission_flags,
-                                      const CService& addr_bind,
-                                      const CService& addr);
+    virtual bool EventNewConnectionAccepted(SockMan::Id id,
+                                            const CService& me,
+                                            const CService& them) override;
+
+    /**
+     * Mark a node as disconnected and close its connection with the peer.
+     * @param[in] node Node to disconnect.
+     */
+    void MarkAsDisconnectAndCloseConnection(CNode& node);
 
     void DisconnectNodes() EXCLUSIVE_LOCKS_REQUIRED(!m_reconnections_mutex, !m_nodes_mutex);
     void NotifyNumConnectionsChanged();
     /** Return true if the peer is inactive and should be disconnected. */
     bool InactivityCheck(const CNode& node) const;
 
-    /**
-     * Generate a collection of sockets to check for IO readiness.
-     * @param[in] nodes Select from these nodes' sockets.
-     * @return sockets to check for readiness
-     */
-    Sock::EventsPerSock GenerateWaitSockets(Span<CNode* const> nodes);
+    void EventReadyToSend(SockMan::Id id, bool& cancel_recv) override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex);
 
-    /**
-     * Check connected and listening sockets for IO readiness and process them accordingly.
-     */
-    void SocketHandler() EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex, !mutexMsgProc);
+    virtual void EventGotData(SockMan::Id id, std::span<const uint8_t> data) override
+        EXCLUSIVE_LOCKS_REQUIRED(!mutexMsgProc, !m_nodes_mutex);
 
-    /**
-     * Do the read/write for connected sockets that are ready for IO.
-     * @param[in] nodes Nodes to process. The socket of each node is checked against `what`.
-     * @param[in] events_per_sock Sockets that are ready for IO.
-     */
-    void SocketHandlerConnected(const std::vector<CNode*>& nodes,
-                                const Sock::EventsPerSock& events_per_sock)
-        EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex, !mutexMsgProc);
+    virtual void EventGotEOF(SockMan::Id id) override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex);
 
-    /**
-     * Accept incoming connections, one from each read-ready listening socket.
-     * @param[in] events_per_sock Sockets that are ready for IO.
-     */
-    void SocketHandlerListening(const Sock::EventsPerSock& events_per_sock);
+    virtual void EventGotPermanentReadError(SockMan::Id id, const std::string& errmsg) override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex);
+
+    virtual bool ShouldTryToSend(SockMan::Id id) const override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex);
+
+    virtual bool ShouldTryToRecv(SockMan::Id id) const override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex);
+
+    virtual void EventIOLoopCompletedForOne(SockMan::Id id) override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex);
+
+    virtual void EventIOLoopCompletedForAll() override
+        EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex, !m_reconnections_mutex);
 
-    void ThreadSocketHandler() EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex, !mutexMsgProc, !m_nodes_mutex, !m_reconnections_mutex);
     void ThreadDNSAddressSeed() EXCLUSIVE_LOCKS_REQUIRED(!m_addr_fetches_mutex, !m_nodes_mutex);
 
     uint64_t CalculateKeyedNetGroup(const CNetAddr& ad) const;
@@ -1363,15 +1327,14 @@ private:
     bool AlreadyConnectedToAddress(const CAddress& addr);
 
     bool AttemptToEvictConnection();
-    CNode* ConnectNode(CAddress addrConnect, const char *pszDest, bool fCountFailure, ConnectionType conn_type, bool use_v2transport) EXCLUSIVE_LOCKS_REQUIRED(!m_unused_i2p_sessions_mutex);
+    CNode* ConnectNode(CAddress addrConnect, const char *pszDest, bool fCountFailure, ConnectionType conn_type, bool use_v2transport);
     void AddWhitelistPermissionFlags(NetPermissionFlags& flags, const CNetAddr &addr, const std::vector<NetWhitelistPermissions>& ranges) const;
 
     void DeleteNode(CNode* pnode);
 
-    NodeId GetNewNodeId();
-
     /** (Try to) send data from node's vSendMsg. Returns (bytes_sent, data_left). */
-    std::pair<size_t, bool> SocketSendData(CNode& node) const EXCLUSIVE_LOCKS_REQUIRED(node.cs_vSend);
+    std::pair<size_t, bool> SendMessagesAsBytes(CNode& node) EXCLUSIVE_LOCKS_REQUIRED(node.cs_vSend)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);
 
     void DumpAddresses();
 
@@ -1430,7 +1393,11 @@ private:
     unsigned int nSendBufferMaxSize{0};
     unsigned int nReceiveFloodSize{0};
 
-    std::vector<ListenSocket> vhListenSocket;
+    /**
+     * Permissions that incoming peers get based on our listening address they connected to.
+     */
+    std::unordered_map<CService, NetPermissionFlags, CServiceHash> m_listen_permissions;
+
     std::atomic<bool> fNetworkActive{true};
     bool fAddressesInitialized{false};
     AddrMan& addrman;
@@ -1441,11 +1408,12 @@ private:
     // connection string and whether to use v2 p2p
     std::vector<AddedNodeParams> m_added_node_params GUARDED_BY(m_added_nodes_mutex);
 
+    CNode* GetNodeById(NodeId node_id) const EXCLUSIVE_LOCKS_REQUIRED(!m_nodes_mutex);
+
     mutable Mutex m_added_nodes_mutex;
-    std::vector<CNode*> m_nodes GUARDED_BY(m_nodes_mutex);
+    std::unordered_map<NodeId, CNode*> m_nodes GUARDED_BY(m_nodes_mutex);
     std::list<CNode*> m_nodes_disconnected;
     mutable RecursiveMutex m_nodes_mutex;
-    std::atomic<NodeId> nLastNodeId{0};
     unsigned int nPrevNodeCount{0};
 
     // Stores number of full-tx connections (outbound and manual) per network
@@ -1540,27 +1508,10 @@ private:
     Mutex mutexMsgProc;
     std::atomic<bool> flagInterruptMsgProc{false};
 
-    /**
-     * This is signaled when network activity should cease.
-     * A pointer to it is saved in `m_i2p_sam_session`, so make sure that
-     * the lifetime of `interruptNet` is not shorter than
-     * the lifetime of `m_i2p_sam_session`.
-     */
-    CThreadInterrupt interruptNet;
-
-    /**
-     * I2P SAM session.
-     * Used to accept incoming and make outgoing I2P connections from a persistent
-     * address.
-     */
-    std::unique_ptr<i2p::sam::Session> m_i2p_sam_session;
-
     std::thread threadDNSAddressSeed;
-    std::thread threadSocketHandler;
     std::thread threadOpenAddedConnections;
     std::thread threadOpenConnections;
     std::thread threadMessageHandler;
-    std::thread threadI2PAcceptIncoming;
 
     /** flag for deciding to connect to an extra outbound peer,
      *  in excess of m_max_outbound_full_relay
@@ -1591,20 +1542,6 @@ private:
      */
     bool whitelist_relay;
 
-    /**
-     * Mutex protecting m_i2p_sam_sessions.
-     */
-    Mutex m_unused_i2p_sessions_mutex;
-
-    /**
-     * A pool of created I2P SAM transient sessions that should be used instead
-     * of creating new ones in order to reduce the load on the I2P network.
-     * Creating a session in I2P is not cheap, thus if this is not empty, then
-     * pick an entry from it instead of creating a new session. If connecting to
-     * a host fails, then the created session is put to this pool for reuse.
-     */
-    std::queue<std::unique_ptr<i2p::sam::Session>> m_unused_i2p_sessions GUARDED_BY(m_unused_i2p_sessions_mutex);
-
     /**
      * Mutex protecting m_reconnections.
      */
@@ -1626,13 +1563,7 @@ private:
     std::list<ReconnectionInfo> m_reconnections GUARDED_BY(m_reconnections_mutex);
 
     /** Attempt reconnections, if m_reconnections non-empty. */
-    void PerformReconnections() EXCLUSIVE_LOCKS_REQUIRED(!m_reconnections_mutex, !m_unused_i2p_sessions_mutex);
-
-    /**
-     * Cap on the size of `m_unused_i2p_sessions`, to ensure it does not
-     * unexpectedly use too much memory.
-     */
-    static constexpr size_t MAX_UNUSED_I2P_SESSIONS_SIZE{10};
+    void PerformReconnections() EXCLUSIVE_LOCKS_REQUIRED(!m_reconnections_mutex);
 
     /**
      * RAII helper to atomically create a copy of `m_nodes` and add a reference
@@ -1645,8 +1576,9 @@ private:
         {
             {
                 LOCK(connman.m_nodes_mutex);
-                m_nodes_copy = connman.m_nodes;
-                for (auto& node : m_nodes_copy) {
+                m_nodes_copy.reserve(connman.m_nodes.size());
+                for (auto& [_, node] : connman.m_nodes) {
+                    m_nodes_copy.push_back(node);
                     node->AddRef();
                 }
             }

src/net_processing.cpp

@@ -5077,10 +5077,15 @@ void PeerManagerImpl::EvictExtraOutboundPeers(std::chrono::seconds now)
 
         m_connman.ForEachNode([&](CNode* pnode) {
             if (!pnode->IsBlockOnlyConn() || pnode->fDisconnect) return;
-            if (pnode->GetId() > youngest_peer.first) {
-                next_youngest_peer = youngest_peer;
-                youngest_peer.first = pnode->GetId();
-                youngest_peer.second = pnode->m_last_block_time;
+            if (pnode->GetId() > next_youngest_peer.first) {
+                if (pnode->GetId() > youngest_peer.first) {
+                    next_youngest_peer = youngest_peer;
+                    youngest_peer.first = pnode->GetId();
+                    youngest_peer.second = pnode->m_last_block_time;
+                } else {
+                    next_youngest_peer.first = pnode->GetId();
+                    next_youngest_peer.second = pnode->m_last_block_time;
+                }
             }
         });
         NodeId to_disconnect = youngest_peer.first;

src/node/txreconciliation.cpp

@@ -87,7 +87,7 @@ public:
         LogPrintLevel(BCLog::TXRECONCILIATION, BCLog::Level::Debug, "Pre-register peer=%d\n", peer_id);
         const uint64_t local_salt{FastRandomContext().rand64()};
 
-        // We do this exactly once per peer (which are unique by NodeId, see GetNewNodeId) so it's
+        // We do this exactly once per peer (which are unique by id, see SockMan::GetNewId()) so it's
         // safe to assume we don't have this record yet.
         Assume(m_states.emplace(peer_id, local_salt).second);
         return local_salt;

src/rest.cpp

@@ -37,6 +37,7 @@
 
 #include <univalue.h>
 
+using http_bitcoin::HTTPRequest;
 using node::GetTransaction;
 using node::NodeContext;
 using util::SplitString;

src/rpc/net.cpp

@@ -203,6 +203,10 @@ static RPCHelpMan getpeerinfo()
     std::vector<CNodeStats> vstats;
     connman.GetNodeStats(vstats);
 
+    std::sort(vstats.begin(), vstats.end(), [](const CNodeStats& a, const CNodeStats& b) {
+        return a.nodeid < b.nodeid;
+    });
+
     UniValue ret(UniValue::VARR);
 
     for (const CNodeStats& stats : vstats) {

src/rpc/node.cpp

@@ -240,24 +240,16 @@ static RPCHelpMan logging()
                 },
                 RPCExamples{
                     HelpExampleCli("logging", "\"[\\\"all\\\"]\" \"[\\\"http\\\"]\"")
-            + HelpExampleRpc("logging", "[\"all\"], [\"libevent\"]")
+            + HelpExampleRpc("logging", "[\"all\"], [\"walletdb\"]")
                 },
         [&](const RPCHelpMan& self, const JSONRPCRequest& request) -> UniValue
 {
-    BCLog::CategoryMask original_log_categories = LogInstance().GetCategoryMask();
     if (request.params[0].isArray()) {
         EnableOrDisableLogCategories(request.params[0], true);
     }
     if (request.params[1].isArray()) {
         EnableOrDisableLogCategories(request.params[1], false);
     }
-    BCLog::CategoryMask updated_log_categories = LogInstance().GetCategoryMask();
-    BCLog::CategoryMask changed_log_categories = original_log_categories ^ updated_log_categories;
-
-    // Update libevent logging if BCLog::LIBEVENT has changed.
-    if (changed_log_categories & BCLog::LIBEVENT) {
-        UpdateHTTPServerLogging(LogInstance().WillLogCategory(BCLog::LIBEVENT));
-    }
 
     UniValue result(UniValue::VOBJ);
     for (const auto& logCatActive : LogInstance().LogCategoriesList()) {

src/rpc/protocol.h

@@ -20,6 +20,20 @@ enum HTTPStatusCode
     HTTP_SERVICE_UNAVAILABLE   = 503,
 };
 
+// Copied from libevent http.c success_phrases[] and client_error_phrases[]
+// TODO: Should HTTPStatusCode and HTTPReason be moved since they are not RPC protocols?
+const std::map<HTTPStatusCode, std::string> HTTPReason{
+    {HTTP_OK, "OK"},
+    {HTTP_NO_CONTENT, "No Content"},
+    {HTTP_BAD_REQUEST, "Bad Request"},
+    {HTTP_UNAUTHORIZED, "Unauthorized"},
+    {HTTP_FORBIDDEN, "Forbidden"},
+    {HTTP_NOT_FOUND, "Not Found"},
+    {HTTP_BAD_METHOD, "Method Not Allowed"},
+    {HTTP_INTERNAL_SERVER_ERROR, "Internal Server Error"},
+    {HTTP_SERVICE_UNAVAILABLE, "Service Unavailable"},
+};
+
 //! Bitcoin RPC error codes
 enum RPCErrorCode
 {

src/test/denialofservice_tests.cpp

@@ -55,7 +55,6 @@ BOOST_AUTO_TEST_CASE(outbound_slow_chain_eviction)
     CAddress addr1(ip(0xa0b0c001), NODE_NONE);
     NodeId id{0};
     CNode dummyNode1{id++,
-                     /*sock=*/nullptr,
                      addr1,
                      /*nKeyedNetGroupIn=*/0,
                      /*nLocalHostNonceIn=*/0,
@@ -121,7 +120,6 @@ void AddRandomOutboundPeer(NodeId& id, std::vector<CNode*>& vNodes, PeerManager&
     }
 
     vNodes.emplace_back(new CNode{id++,
-                                  /*sock=*/nullptr,
                                   addr,
                                   /*nKeyedNetGroupIn=*/0,
                                   /*nLocalHostNonceIn=*/0,
@@ -320,7 +318,6 @@ BOOST_AUTO_TEST_CASE(peer_discouragement)
     banman->ClearBanned();
     NodeId id{0};
     nodes[0] = new CNode{id++,
-                         /*sock=*/nullptr,
                          addr[0],
                          /*nKeyedNetGroupIn=*/0,
                          /*nLocalHostNonceIn=*/0,
@@ -340,7 +337,6 @@ BOOST_AUTO_TEST_CASE(peer_discouragement)
     BOOST_CHECK(!banman->IsDiscouraged(other_addr)); // Different address, not discouraged
 
     nodes[1] = new CNode{id++,
-                         /*sock=*/nullptr,
                          addr[1],
                          /*nKeyedNetGroupIn=*/1,
                          /*nLocalHostNonceIn=*/1,
@@ -370,7 +366,6 @@ BOOST_AUTO_TEST_CASE(peer_discouragement)
     // Make sure non-IP peers are discouraged and disconnected properly.
 
     nodes[2] = new CNode{id++,
-                         /*sock=*/nullptr,
                          addr[2],
                          /*nKeyedNetGroupIn=*/1,
                          /*nLocalHostNonceIn=*/1,
@@ -412,7 +407,6 @@ BOOST_AUTO_TEST_CASE(DoS_bantime)
     CAddress addr(ip(0xa0b0c001), NODE_NONE);
     NodeId id{0};
     CNode dummyNode{id++,
-                    /*sock=*/nullptr,
                     addr,
                     /*nKeyedNetGroupIn=*/4,
                     /*nLocalHostNonceIn=*/4,

src/test/fuzz/connman.cpp

@@ -64,13 +64,15 @@ FUZZ_TARGET(connman, .init = initialize_connman)
     connman.Init(options);
 
     CNetAddr random_netaddr;
-    CNode random_node = ConsumeNode(fuzzed_data_provider);
+    NodeId node_id{0};
+    CNode& random_node{*ConsumeNodeAsUniquePtr(fuzzed_data_provider, node_id++).release()};
+    connman.AddTestNode(random_node, std::make_unique<FuzzedSock>(fuzzed_data_provider));
     CSubNet random_subnet;
     std::string random_string;
 
     LIMITED_WHILE(fuzzed_data_provider.ConsumeBool(), 100) {
-        CNode& p2p_node{*ConsumeNodeAsUniquePtr(fuzzed_data_provider).release()};
-        connman.AddTestNode(p2p_node);
+        CNode& p2p_node{*ConsumeNodeAsUniquePtr(fuzzed_data_provider, node_id++).release()};
+        connman.AddTestNode(p2p_node, std::make_unique<FuzzedSock>(fuzzed_data_provider));
     }
 
     LIMITED_WHILE(fuzzed_data_provider.ConsumeBool(), 10000) {
@@ -103,6 +105,15 @@ FUZZ_TARGET(connman, .init = initialize_connman)
             [&] {
                 connman.DisconnectNode(random_subnet);
             },
+            [&] {
+                if (fuzzed_data_provider.ConsumeBool()) {
+                    auto nonexistent_node{ConsumeNodeAsUniquePtr(fuzzed_data_provider, node_id++)};
+                    connman.MarkAsDisconnectAndCloseConnection(*nonexistent_node);
+                } else {
+                    CNode& existent_node{*connman.TestNodes().begin()->second};
+                    connman.MarkAsDisconnectAndCloseConnection(existent_node);
+                }
+            },
             [&] {
                 connman.ForEachNode([](auto) {});
             },

src/test/fuzz/http_request.cpp

@@ -10,47 +10,25 @@
 #include <util/signalinterrupt.h>
 #include <util/strencodings.h>
 
-#include <event2/buffer.h>
-#include <event2/event.h>
-#include <event2/http.h>
-#include <event2/http_struct.h>
-
 #include <cassert>
 #include <cstdint>
 #include <string>
 #include <vector>
 
+using http_bitcoin::HTTPRequest;
+
 extern "C" int evhttp_parse_firstline_(struct evhttp_request*, struct evbuffer*);
 extern "C" int evhttp_parse_headers_(struct evhttp_request*, struct evbuffer*);
 
-std::string RequestMethodString(HTTPRequest::RequestMethod m);
+std::string RequestMethodString(HTTPRequestMethod m);
 
 FUZZ_TARGET(http_request)
 {
     FuzzedDataProvider fuzzed_data_provider{buffer.data(), buffer.size()};
-    evhttp_request* evreq = evhttp_request_new(nullptr, nullptr);
-    assert(evreq != nullptr);
-    evreq->kind = EVHTTP_REQUEST;
-    evbuffer* evbuf = evbuffer_new();
-    assert(evbuf != nullptr);
-    const std::vector<uint8_t> http_buffer = ConsumeRandomLengthByteVector(fuzzed_data_provider, 4096);
-    evbuffer_add(evbuf, http_buffer.data(), http_buffer.size());
-    // Avoid constructing requests that will be interpreted by libevent as PROXY requests to avoid triggering
-    // a nullptr dereference. The dereference (req->evcon->http_server) takes place in evhttp_parse_request_line
-    // and is a consequence of our hacky but necessary use of the internal function evhttp_parse_firstline_ in
-    // this fuzzing harness. The workaround is not aesthetically pleasing, but it successfully avoids the troublesome
-    // code path. " http:// HTTP/1.1\n" was a crashing input prior to this workaround.
-    const std::string http_buffer_str = ToLower(std::string{http_buffer.begin(), http_buffer.end()});
-    if (http_buffer_str.find(" http://") != std::string::npos || http_buffer_str.find(" https://") != std::string::npos ||
-        evhttp_parse_firstline_(evreq, evbuf) != 1 || evhttp_parse_headers_(evreq, evbuf) != 1) {
-        evbuffer_free(evbuf);
-        evhttp_request_free(evreq);
-        return;
-    }
 
     util::SignalInterrupt interrupt;
-    HTTPRequest http_request{evreq, interrupt, true};
-    const HTTPRequest::RequestMethod request_method = http_request.GetRequestMethod();
+    HTTPRequest http_request;
+    const HTTPRequestMethod request_method = http_request.GetRequestMethod();
     (void)RequestMethodString(request_method);
     (void)http_request.GetURI();
     (void)http_request.GetHeader("Host");
@@ -62,7 +40,4 @@ FUZZ_TARGET(http_request)
     assert(body.empty());
     const CService service = http_request.GetPeer();
     assert(service.ToStringAddrPort() == "[::]:0");
-
-    evbuffer_free(evbuf);
-    evhttp_request_free(evreq);
 }

src/test/fuzz/net.cpp

@@ -42,9 +42,6 @@ FUZZ_TARGET(net, .init = initialize_net)
     LIMITED_WHILE(fuzzed_data_provider.ConsumeBool(), 10000) {
         CallOneOf(
             fuzzed_data_provider,
-            [&] {
-                node.CloseSocketDisconnect();
-            },
             [&] {
                 CNodeStats stats;
                 node.CopyStats(stats);

src/test/fuzz/p2p_handshake.cpp

@@ -65,7 +65,7 @@ FUZZ_TARGET(p2p_handshake, .init = ::initialize)
     const auto num_peers_to_add = fuzzed_data_provider.ConsumeIntegralInRange(1, 3);
     for (int i = 0; i < num_peers_to_add; ++i) {
         peers.push_back(ConsumeNodeAsUniquePtr(fuzzed_data_provider, i).release());
-        connman.AddTestNode(*peers.back());
+        connman.AddTestNode(*peers.back(), std::make_unique<FuzzedSock>(fuzzed_data_provider));
         peerman->InitializeNode(
             *peers.back(),
             static_cast<ServiceFlags>(fuzzed_data_provider.ConsumeIntegral<uint64_t>()));

src/test/fuzz/p2p_headers_presync.cpp

@@ -60,7 +60,7 @@ void HeadersSyncSetup::ResetAndInitialize()
 
     for (auto conn_type : conn_types) {
         CAddress addr{};
-        m_connections.push_back(new CNode(id++, nullptr, addr, 0, 0, addr, "", conn_type, false));
+        m_connections.push_back(new CNode(id++, addr, 0, 0, addr, "", conn_type, false));
         CNode& p2p_node = *m_connections.back();
 
         connman.Handshake(

src/test/fuzz/process_message.cpp

@@ -68,7 +68,7 @@ FUZZ_TARGET(process_message, .init = initialize_process_message)
     }
     CNode& p2p_node = *ConsumeNodeAsUniquePtr(fuzzed_data_provider).release();
 
-    connman.AddTestNode(p2p_node);
+    connman.AddTestNode(p2p_node, std::make_unique<FuzzedSock>(fuzzed_data_provider));
     FillNode(fuzzed_data_provider, connman, p2p_node);
 
     const auto mock_time = ConsumeTime(fuzzed_data_provider);

src/test/fuzz/process_messages.cpp

@@ -60,7 +60,7 @@ FUZZ_TARGET(process_messages, .init = initialize_process_messages)
 
         FillNode(fuzzed_data_provider, connman, p2p_node);
 
-        connman.AddTestNode(p2p_node);
+        connman.AddTestNode(p2p_node, std::make_unique<FuzzedSock>(fuzzed_data_provider));
     }
 
     LIMITED_WHILE(fuzzed_data_provider.ConsumeBool(), 30)

src/test/fuzz/util/net.h

@@ -231,7 +231,6 @@ template <bool ReturnUniquePtr = false>
 auto ConsumeNode(FuzzedDataProvider& fuzzed_data_provider, const std::optional<NodeId>& node_id_in = std::nullopt) noexcept
 {
     const NodeId node_id = node_id_in.value_or(fuzzed_data_provider.ConsumeIntegralInRange<NodeId>(0, std::numeric_limits<NodeId>::max()));
-    const auto sock = std::make_shared<FuzzedSock>(fuzzed_data_provider);
     const CAddress address = ConsumeAddress(fuzzed_data_provider);
     const uint64_t keyed_net_group = fuzzed_data_provider.ConsumeIntegral<uint64_t>();
     const uint64_t local_host_nonce = fuzzed_data_provider.ConsumeIntegral<uint64_t>();
@@ -242,7 +241,6 @@ auto ConsumeNode(FuzzedDataProvider& fuzzed_data_provider, const std::optional<N
     NetPermissionFlags permission_flags = ConsumeWeakEnum(fuzzed_data_provider, ALL_NET_PERMISSION_FLAGS);
     if constexpr (ReturnUniquePtr) {
         return std::make_unique<CNode>(node_id,
-                                       sock,
                                        address,
                                        keyed_net_group,
                                        local_host_nonce,
@@ -253,7 +251,6 @@ auto ConsumeNode(FuzzedDataProvider& fuzzed_data_provider, const std::optional<N
                                        CNodeOptions{ .permission_flags = permission_flags });
     } else {
         return CNode{node_id,
-                     sock,
                      address,
                      keyed_net_group,
                      local_host_nonce,

src/test/httpserver_tests.cpp

@@ -3,14 +3,81 @@
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
 #include <httpserver.h>
+#include <time.h>
+#include <rpc/protocol.h>
+#include <test/util/net.h>
 #include <test/util/setup_common.h>
+#include <util/strencodings.h>
 
 #include <boost/test/unit_test.hpp>
 
-BOOST_FIXTURE_TEST_SUITE(httpserver_tests, BasicTestingSetup)
+using http_bitcoin::HTTPClient;
+using http_bitcoin::HTTPHeaders;
+using http_bitcoin::HTTPRequest;
+using http_bitcoin::HTTPResponse;
+using http_bitcoin::HTTPServer;
+using http_bitcoin::MAX_HEADERS_SIZE;
+using util::LineReader;
+
+// Reading request captured from bitcoin-cli
+const std::string full_request =
+    "504f5354202f20485454502f312e310d0a486f73743a203132372e302e302e310d"
+    "0a436f6e6e656374696f6e3a20636c6f73650d0a436f6e74656e742d547970653a"
+    "206170706c69636174696f6e2f6a736f6e0d0a417574686f72697a6174696f6e3a"
+    "204261736963205831396a6232397261575666587a6f354f4751354f4451334d57"
+    "4e6d4e6a67304e7a417a59546b7a4e32457a4e7a6b305a44466c4f4451314e6a5a"
+    "6d5954526b5a6a4a694d7a466b596a68684f4449345a4759344d6a566a4f546735"
+    "5a4749344f54566c0d0a436f6e74656e742d4c656e6774683a2034360d0a0d0a7b"
+    "226d6574686f64223a22676574626c6f636b636f756e74222c22706172616d7322"
+    "3a5b5d2c226964223a317d0a";
+
+/// Save the value of CreateSock and restore it when the test ends.
+class HTTPTestingSetup : public BasicTestingSetup
+{
+public:
+    explicit HTTPTestingSetup() : m_create_sock_orig{CreateSock} {};
+
+    ~HTTPTestingSetup()
+    {
+        CreateSock = m_create_sock_orig;
+    }
+
+private:
+    const decltype(CreateSock) m_create_sock_orig;
+};
+
+BOOST_FIXTURE_TEST_SUITE(httpserver_tests, HTTPTestingSetup)
 
 BOOST_AUTO_TEST_CASE(test_query_parameters)
 {
+    // The legacy code that relied on libevent couldn't handle an invalid URI encoding.
+    // The new code is more tolerant and so we expect a difference in behavior.
+    // Re: libevent evhttp_uri_parse() see:
+    //   "bugfix: rest: avoid segfault for invalid URI" https://github.com/bitcoin/bitcoin/pull/27468
+    //   "httpserver, rest: improving URI validation" https://github.com/bitcoin/bitcoin/pull/27253
+    // Re: More tolerant URI decoding see:
+    //   "refactor: Use our own implementation of urlDecode" https://github.com/bitcoin/bitcoin/pull/29904
+
+    std::string uri {};
+    // This is an invalid URI because it contains a % that is not followed by two hex digits
+    uri = "/rest/endpoint/someresource.json?p1=v1&p2=v2%";
+    // Old libevent behavior: URI with invalid characters (%) raised a runtime error regardless of which query parameter is queried
+    // New behavior: Tolerate as much as we can even
+    BOOST_CHECK_EQUAL(http_bitcoin::GetQueryParameterFromUri(uri.c_str(), "p1").value(), "v1");
+    BOOST_CHECK_EQUAL(http_bitcoin::GetQueryParameterFromUri(uri.c_str(), "p2").value(), "v2%");
+
+    // This is a valid URI because the %XX encoding is correct: `?p1=v1&p2=100%`
+    uri = "/rest/endpoint/someresource.json%3Fp1%3Dv1%26p2%3D100%25";
+    // Old behavior: libevent did not decode the URI before parsing, so it did not detect or return the query
+    //               (libevent would parse the entire argument string as the uri path)
+    // New behavior: Decode before parsing the URI so reserved characters like ? & = are interpreted correctly
+    BOOST_CHECK_EQUAL(http_bitcoin::GetQueryParameterFromUri(uri.c_str(), "p1").value(), "v1");
+    BOOST_CHECK_EQUAL(http_bitcoin::GetQueryParameterFromUri(uri.c_str(), "p2").value(), "100%");
+}
+
+// Ensure new behavior matches old behavior
+template <typename func>
+void test_query_parameters(func GetQueryParameterFromUri) {
     std::string uri {};
 
     // No parameters
@@ -35,8 +102,341 @@ BOOST_AUTO_TEST_CASE(test_query_parameters)
     uri = "/rest/endpoint/someresource.json&p1=v1&p2=v2";
     BOOST_CHECK(!GetQueryParameterFromUri(uri.c_str(), "p1").has_value());
 
-    // URI with invalid characters (%) raises a runtime error regardless of which query parameter is queried
-    uri = "/rest/endpoint/someresource.json&p1=v1&p2=v2%";
-    BOOST_CHECK_EXCEPTION(GetQueryParameterFromUri(uri.c_str(), "p1"), std::runtime_error, HasReason("URI parsing failed, it likely contained RFC 3986 invalid characters"));
+    // Multiple parameters, some characters encoded
+    uri = "/rest/endpoint/someresource.json?p1=v1%20&p2=100%25";
+    BOOST_CHECK_EQUAL(GetQueryParameterFromUri(uri.c_str(), "p1").value(), "v1 ");
+    BOOST_CHECK_EQUAL(GetQueryParameterFromUri(uri.c_str(), "p2").value(), "100%");
+}
+
+BOOST_AUTO_TEST_CASE(test_query_parameters_bitcoin)
+{
+    test_query_parameters(http_bitcoin::GetQueryParameterFromUri);
+}
+
+BOOST_AUTO_TEST_CASE(http_headers_tests)
+{
+    {
+        // Writing response headers
+        HTTPHeaders headers{};
+        BOOST_CHECK(!headers.Find("Cache-Control"));
+        headers.Write("Cache-Control", "no-cache");
+        // Check case-insensitive key matching
+        BOOST_CHECK_EQUAL(headers.Find("Cache-Control").value(), "no-cache");
+        BOOST_CHECK_EQUAL(headers.Find("cache-control").value(), "no-cache");
+        // Additional values are comma-separated and appended
+        headers.Write("Cache-Control", "no-store");
+        BOOST_CHECK_EQUAL(headers.Find("Cache-Control").value(), "no-cache, no-store");
+        // Add a few more
+        headers.Write("Pie", "apple");
+        headers.Write("Sandwich", "ham");
+        headers.Write("Coffee", "black");
+        BOOST_CHECK_EQUAL(headers.Find("Pie").value(), "apple");
+        // Remove
+        headers.Remove("Pie");
+        BOOST_CHECK(!headers.Find("Pie"));
+        // Combine for transmission
+        // std::map sorts alphabetically by key, no order is specified for HTTP
+        BOOST_CHECK_EQUAL(
+            headers.Stringify(),
+            "Cache-Control: no-cache, no-store\r\n"
+            "Coffee: black\r\n"
+            "Sandwich: ham\r\n\r\n");
+    }
+    {
+        // Reading request headers captured from bitcoin-cli
+        std::vector<std::byte> buffer{TryParseHex<std::byte>(
+            "486f73743a203132372e302e302e310d0a436f6e6e656374696f6e3a20636c6f73"
+            "650d0a436f6e74656e742d547970653a206170706c69636174696f6e2f6a736f6e"
+            "0d0a417574686f72697a6174696f6e3a204261736963205831396a623239726157"
+            "5666587a6f7a597a4a6b4e5441784e44466c4d474a69596d56684d5449354f4467"
+            "334e7a49354d544d334e54526d4e54686b4e6a63324f574d775a5459785a6a677a"
+            "4e5467794e7a4577595459314f47526b596a566d5a4751330d0a436f6e74656e74"
+            "2d4c656e6774683a2034360d0a0d0a").value()};
+        util::LineReader reader(buffer, /*max_read=*/1028);
+        HTTPHeaders headers{};
+        headers.Read(reader);
+        BOOST_CHECK_EQUAL(headers.Find("Host").value(), "127.0.0.1");
+        BOOST_CHECK_EQUAL(headers.Find("Connection").value(), "close");
+        BOOST_CHECK_EQUAL(headers.Find("Content-Type").value(), "application/json");
+        BOOST_CHECK_EQUAL(headers.Find("Authorization").value(), "Basic X19jb29raWVfXzozYzJkNTAxNDFlMGJiYmVhMTI5ODg3NzI5MTM3NTRmNThkNjc2OWMwZTYxZjgzNTgyNzEwYTY1OGRkYjVmZGQ3");
+        BOOST_CHECK_EQUAL(headers.Find("Content-Length").value(), "46");
+        BOOST_CHECK(!headers.Find("Pizza"));
+    }
+}
+
+BOOST_AUTO_TEST_CASE(http_response_tests)
+{
+    // Typical HTTP 1.1 response headers
+    HTTPHeaders headers{};
+    headers.Write("Content-Type", "application/json");
+    headers.Write("Date", "Tue, 15 Oct 2024 17:54:12 GMT");
+    headers.Write("Content-Length", "41");
+    // Response points to headers which already exist because some of them
+    // are set before we even know what the response will be.
+    HTTPResponse res;
+    res.m_version_major = 1;
+    res.m_version_minor = 1;
+    res.m_status = HTTP_OK;
+    res.m_reason = HTTPReason.find(res.m_status)->second;
+    res.m_body = StringToBuffer("{\"result\":865793,\"error\":null,\"id\":null\"}");
+    // Everything except the body, which might be raw bytes instead of a string
+    res.m_headers = std::move(headers);
+    BOOST_CHECK_EQUAL(
+        res.StringifyHeaders(),
+        "HTTP/1.1 200 OK\r\n"
+        "Content-Length: 41\r\n"
+        "Content-Type: application/json\r\n"
+        "Date: Tue, 15 Oct 2024 17:54:12 GMT\r\n"
+        "\r\n");
+}
+
+BOOST_AUTO_TEST_CASE(http_request_tests)
+{
+    {
+        HTTPRequest req;
+        std::vector<std::byte> buffer{TryParseHex<std::byte>(full_request).value()};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK(req.LoadControlData(reader));
+        BOOST_CHECK(req.LoadHeaders(reader));
+        BOOST_CHECK(req.LoadBody(reader));
+        BOOST_CHECK_EQUAL(req.m_method, "POST");
+        BOOST_CHECK_EQUAL(req.GetRequestMethod(), HTTPRequestMethod::POST);
+        BOOST_CHECK_EQUAL(req.m_target, "/");
+        BOOST_CHECK_EQUAL(req.GetURI(), "/");
+        BOOST_CHECK_EQUAL(req.m_version_major, 1);
+        BOOST_CHECK_EQUAL(req.m_version_minor, 1);
+        BOOST_CHECK_EQUAL(req.m_headers.Find("Host").value(), "127.0.0.1");
+        BOOST_CHECK_EQUAL(req.m_headers.Find("Connection").value(), "close");
+        BOOST_CHECK_EQUAL(req.m_headers.Find("Content-Type").value(), "application/json");
+        BOOST_CHECK_EQUAL(req.m_headers.Find("Authorization").value(), "Basic X19jb29raWVfXzo5OGQ5ODQ3MWNmNjg0NzAzYTkzN2EzNzk0ZDFlODQ1NjZmYTRkZjJiMzFkYjhhODI4ZGY4MjVjOTg5ZGI4OTVl");
+        BOOST_CHECK_EQUAL(req.m_headers.Find("Content-Length").value(), "46");
+        BOOST_CHECK_EQUAL(req.m_body.size(), 46);
+        BOOST_CHECK_EQUAL(req.m_body, "{\"method\":\"getblockcount\",\"params\":[],\"id\":1}\n");
+    }
+    {
+        const std::string too_short_request_line = "GET/HTTP/1.0\r\nHost: 127.0.0.1\r\n\r\n";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(too_short_request_line)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK_THROW(req.LoadControlData(reader), std::runtime_error);
+    }
+    {
+        const std::string malformed_request_line = "GET / HTTP / 1.0\r\nHost: 127.0.0.1\r\n\r\n";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(malformed_request_line)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK_THROW(req.LoadControlData(reader), std::runtime_error);
+    }
+    {
+        const std::string malformed_request_line = "GET / HTTP1.0\r\nHost: 127.0.0.1\r\n\r\n";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(malformed_request_line)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK_THROW(req.LoadControlData(reader), std::runtime_error);
+    }
+    {
+        const std::string malformed_request_line = "GET / HTTP/11\r\nHost: 127.0.0.1\r\n\r\n";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(malformed_request_line)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK_THROW(req.LoadControlData(reader), std::runtime_error);
+    }
+    {
+        const std::string malformed_request_line = "GET / HTTP/1.x\r\nHost: 127.0.0.1\r\n\r\n";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(malformed_request_line)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK_THROW(req.LoadControlData(reader), std::runtime_error);
+    }
+    {
+        const std::string ok_request_line = "GET / HTTP/1.0\r\nHost: 127.0.0.1\r\n\r\n";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(ok_request_line)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK(req.LoadControlData(reader));
+        BOOST_CHECK(req.LoadHeaders(reader));
+        BOOST_CHECK(req.LoadBody(reader));
+        BOOST_CHECK_EQUAL(req.m_method, "GET");
+        BOOST_CHECK_EQUAL(req.m_target, "/");
+        BOOST_CHECK_EQUAL(req.m_version_major, 1);
+        BOOST_CHECK_EQUAL(req.m_version_minor, 0);
+        BOOST_CHECK_EQUAL(req.m_headers.Find("Host").value(), "127.0.0.1");
+        // no body is OK
+        BOOST_CHECK_EQUAL(req.m_body.size(), 0);
+    }
+    {
+        const std::string malformed_headers = "GET / HTTP/1.0\r\nHost=127.0.0.1\r\n\r\n";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(malformed_headers)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK(req.LoadControlData(reader));
+        BOOST_CHECK_THROW(req.LoadHeaders(reader), std::runtime_error);
+    }
+    {
+        // We might not have received enough data from the client which is not
+        // an error. We return false so the caller can try again later when the
+        // buffer has more data.
+        const std::string incomplete_headers = "GET / HTTP/1.0\r\nHost: ";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(incomplete_headers)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK(req.LoadControlData(reader));
+        BOOST_CHECK(!req.LoadHeaders(reader));
+    }
+    {
+        const std::string no_content_length = "GET / HTTP/1.0\r\n\r\n{\"method\":\"getblockcount\"}";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(no_content_length)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK(req.LoadControlData(reader));
+        BOOST_CHECK(req.LoadHeaders(reader));
+        BOOST_CHECK(req.LoadBody(reader));
+        // Don't try to read request body if Content-Length is missing
+        BOOST_CHECK_EQUAL(req.m_body.size(), 0);
+    }
+    {
+        const std::string bad_content_length = "GET / HTTP/1.0\r\nContent-Length: eleven\r\n\r\n{\"method\":\"getblockcount\"}";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(bad_content_length)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK(req.LoadControlData(reader));
+        BOOST_CHECK(req.LoadHeaders(reader));
+        BOOST_CHECK_THROW(req.LoadBody(reader), std::runtime_error);
+    }
+    {
+        // Content-Length indicates more data than we have in the buffer.
+        // Again, not an error just try again later.
+        const std::string excessive_content_length = "GET / HTTP/1.0\r\nContent-Length: 1024\r\n\r\n{\"method\":\"getblockcount\"}";
+        HTTPRequest req;
+        std::vector<std::byte> buffer{StringToBuffer(excessive_content_length)};
+        LineReader reader(buffer, MAX_HEADERS_SIZE);
+        BOOST_CHECK(req.LoadControlData(reader));
+        BOOST_CHECK(req.LoadHeaders(reader));
+        BOOST_CHECK(!req.LoadBody(reader));
+    }
+}
+
+BOOST_AUTO_TEST_CASE(http_client_server_tests)
+{
+    // Hard code the timestamp for the Date header in the HTTP response
+    // Wed Dec 11 00:47:09 2024 UTC
+    SetMockTime(1733878029);
+
+    // Queue of connected sockets returned by listening socket (represents network interface)
+    std::shared_ptr<DynSock::Queue> accepted_sockets{std::make_shared<DynSock::Queue>()};
+
+    CreateSock = [&accepted_sockets](int, int, int) {
+        // This is a mock Listening Socket that the HTTP server will "bind" to and
+        // listen to for incoming connections. We won't need to access its I/O
+        // pipes because we don't read or write directly to it. It will return
+        // Connected Sockets from the queue via its Accept() method.
+        return std::make_unique<DynSock>(std::make_shared<DynSock::Pipes>(), accepted_sockets);
+    };
+
+    {
+        // I/O pipes of one mock Connected Socket we can read and write to.
+        std::shared_ptr<DynSock::Pipes> connected_socket_pipes(std::make_shared<DynSock::Pipes>());
+
+        // Insert the payload: a correctly formatted HTTP request
+        std::vector<std::byte> buffer{TryParseHex<std::byte>(full_request).value()};
+        connected_socket_pipes->recv.PushBytes(buffer.data(), buffer.size());
+
+        // Mock Connected Socket that represents a client.
+        // It needs I/O pipes but its queue can remain empty
+        std::unique_ptr<DynSock> connected_socket{std::make_unique<DynSock>(connected_socket_pipes, std::make_shared<DynSock::Queue>())};
+
+        // Prepare queue of accepted_sockets: just one connection with no data
+        accepted_sockets->Push(std::move(connected_socket));
+
+        // Prepare a request handler that just stores received requests so we can examine them
+        // Mutex is required to prevent a race between this test's main thread and the Sockman I/O loop.
+        Mutex requests_mutex;
+        std::deque<std::unique_ptr<HTTPRequest>> requests;
+        auto StoreRequest = [&](std::unique_ptr<HTTPRequest> req) {
+            LOCK(requests_mutex);
+            requests.push_back(std::move(req));
+        };
+
+        // Instantiate server with dead-end request handler
+        HTTPServer server = HTTPServer(StoreRequest);
+        BOOST_REQUIRE(server.m_no_clients);
+
+        // This address won't actually get used because we stubbed CreateSock()
+        const std::optional<CService> addr{Lookup("127.0.0.1", 8333, false)};
+        bilingual_str strError;
+        // Bind to mock Listening Socket
+        BOOST_REQUIRE(server.BindAndStartListening(addr.value(), strError));
+        // Start the I/O loop, accepting connections
+        SockMan::Options sockman_options;
+        server.StartSocketsThreads(sockman_options);
+
+        // Wait up to one second for mock client to connect.
+        // Given that the mock client is itself a mock socket
+        // with hard-coded data it should only take a fraction of that.
+        int attempts{100};
+        while (attempts > 0)
+        {
+            if (!server.m_no_clients) break;
+
+            std::this_thread::sleep_for(10ms);
+            --attempts;
+        }
+        BOOST_REQUIRE(!server.m_no_clients);
+
+        {
+            LOCK(requests_mutex);
+            // Connected client should have one request already from the static content.
+            BOOST_CHECK_EQUAL(requests.size(), 1);
+
+            // Check the received request
+            BOOST_CHECK_EQUAL(requests.front()->m_body, "{\"method\":\"getblockcount\",\"params\":[],\"id\":1}\n");
+            BOOST_CHECK_EQUAL(requests.front()->GetPeer().ToStringAddrPort(), "5.5.5.5:6789");
+
+            // Respond to request
+            requests.front()->WriteReply(HTTP_OK, "874140\n");
+        }
+
+        // Check the sent response from the mock client at the other end of the mock socket
+        std::string expected = "HTTP/1.1 200 OK\r\n"
+                               "Connection: close\r\n"
+                               "Content-Length: 7\r\n"
+                               "Content-Type: text/html; charset=ISO-8859-1\r\n"
+                               "Date: Wed, 11 Dec 2024 00:47:09 GMT\r\n"
+                               "\r\n"
+                               "874140\n";
+        std::string actual;
+
+        // Wait up to one second for all the bytes to appear in the "send" pipe.
+        char buf[0x10000] = {};
+        attempts = 100;
+        while (attempts > 0)
+        {
+            ssize_t bytes_read = connected_socket_pipes->send.GetBytes(buf, sizeof(buf), 0);
+            if (bytes_read > 0) {
+                actual.append(buf, bytes_read);
+                if (actual == expected) break;
+            }
+
+            std::this_thread::sleep_for(10ms);
+            --attempts;
+        }
+        BOOST_CHECK_EQUAL(actual, expected);
+
+        // Wait up to one second for connection to be closed
+        attempts = 100;
+        while (attempts > 0)
+        {
+            if (server.m_no_clients) break;
+
+            std::this_thread::sleep_for(10ms);
+            --attempts;
+        }
+        BOOST_REQUIRE(server.m_no_clients);
+
+        // Close server
+        server.interruptNet();
+        // Wait for I/O loop to finish, after all sockets are closed
+        server.JoinSocketsThreads();
+    }
 }
 BOOST_AUTO_TEST_SUITE_END()

src/test/net_peer_connection_tests.cpp

@@ -65,7 +65,6 @@ void AddPeer(NodeId& id, std::vector<CNode*>& nodes, PeerManager& peerman, Connm
     const bool inbound_onion{onion_peer && conn_type == ConnectionType::INBOUND};
 
     nodes.emplace_back(new CNode{++id,
-                                 /*sock=*/nullptr,
                                  addr,
                                  /*nKeyedNetGroupIn=*/0,
                                  /*nLocalHostNonceIn=*/0,
@@ -117,9 +116,9 @@ BOOST_FIXTURE_TEST_CASE(test_addnode_getaddednodeinfo_and_connection_detection,
     BOOST_CHECK_EQUAL(nodes.back()->ConnectedThroughNetwork(), Network::NET_CJDNS);
 
     BOOST_TEST_MESSAGE("Call AddNode() for all the peers");
-    for (auto node : connman->TestNodes()) {
+    for (const auto& [id, node] : connman->TestNodes()) {
         BOOST_CHECK(connman->AddNode({/*m_added_node=*/node->addr.ToStringAddrPort(), /*m_use_v2transport=*/true}));
-        BOOST_TEST_MESSAGE(strprintf("peer id=%s addr=%s", node->GetId(), node->addr.ToStringAddrPort()));
+        BOOST_TEST_MESSAGE(strprintf("peer id=%s addr=%s", id, node->addr.ToStringAddrPort()));
     }
 
     BOOST_TEST_MESSAGE("\nCall AddNode() with 2 addrs resolving to existing localhost addnode entry; neither should be added");
@@ -134,7 +133,7 @@ BOOST_FIXTURE_TEST_CASE(test_addnode_getaddednodeinfo_and_connection_detection,
     BOOST_CHECK(connman->GetAddedNodeInfo(/*include_connected=*/false).empty());
 
     // Test AddedNodesContain()
-    for (auto node : connman->TestNodes()) {
+    for (const auto& [_, node] : connman->TestNodes()) {
         BOOST_CHECK(connman->AddedNodesContain(node->addr));
     }
     AddPeer(id, nodes, *peerman, *connman, ConnectionType::OUTBOUND_FULL_RELAY);
@@ -151,12 +150,12 @@ BOOST_FIXTURE_TEST_CASE(test_addnode_getaddednodeinfo_and_connection_detection,
     }
 
     BOOST_TEST_MESSAGE("\nCheck that all connected peers are correctly detected as connected");
-    for (auto node : connman->TestNodes()) {
+    for (const auto& [_, node] : connman->TestNodes()) {
         BOOST_CHECK(connman->AlreadyConnectedPublic(node->addr));
     }
 
     // Clean up
-    for (auto node : connman->TestNodes()) {
+    for (const auto& [_, node] : connman->TestNodes()) {
         peerman->FinalizeNode(*node);
     }
     connman->ClearTestNodes();

src/test/net_tests.cpp

@@ -60,7 +60,6 @@ BOOST_AUTO_TEST_CASE(cnode_simple_test)
     std::string pszDest;
 
     std::unique_ptr<CNode> pnode1 = std::make_unique<CNode>(id++,
-                                                            /*sock=*/nullptr,
                                                             addr,
                                                             /*nKeyedNetGroupIn=*/0,
                                                             /*nLocalHostNonceIn=*/0,
@@ -78,7 +77,6 @@ BOOST_AUTO_TEST_CASE(cnode_simple_test)
     BOOST_CHECK_EQUAL(pnode1->ConnectedThroughNetwork(), Network::NET_IPV4);
 
     std::unique_ptr<CNode> pnode2 = std::make_unique<CNode>(id++,
-                                                            /*sock=*/nullptr,
                                                             addr,
                                                             /*nKeyedNetGroupIn=*/1,
                                                             /*nLocalHostNonceIn=*/1,
@@ -96,7 +94,6 @@ BOOST_AUTO_TEST_CASE(cnode_simple_test)
     BOOST_CHECK_EQUAL(pnode2->ConnectedThroughNetwork(), Network::NET_IPV4);
 
     std::unique_ptr<CNode> pnode3 = std::make_unique<CNode>(id++,
-                                                            /*sock=*/nullptr,
                                                             addr,
                                                             /*nKeyedNetGroupIn=*/0,
                                                             /*nLocalHostNonceIn=*/0,
@@ -114,7 +111,6 @@ BOOST_AUTO_TEST_CASE(cnode_simple_test)
     BOOST_CHECK_EQUAL(pnode3->ConnectedThroughNetwork(), Network::NET_IPV4);
 
     std::unique_ptr<CNode> pnode4 = std::make_unique<CNode>(id++,
-                                                            /*sock=*/nullptr,
                                                             addr,
                                                             /*nKeyedNetGroupIn=*/1,
                                                             /*nLocalHostNonceIn=*/1,
@@ -606,7 +602,6 @@ BOOST_AUTO_TEST_CASE(ipv4_peer_with_ipv6_addrMe_test)
     ipv4AddrPeer.s_addr = 0xa0b0c001;
     CAddress addr = CAddress(CService(ipv4AddrPeer, 7777), NODE_NETWORK);
     std::unique_ptr<CNode> pnode = std::make_unique<CNode>(/*id=*/0,
-                                                           /*sock=*/nullptr,
                                                            addr,
                                                            /*nKeyedNetGroupIn=*/0,
                                                            /*nLocalHostNonceIn=*/0,
@@ -660,7 +655,6 @@ BOOST_AUTO_TEST_CASE(get_local_addr_for_peer_port)
     in_addr peer_out_in_addr;
     peer_out_in_addr.s_addr = htonl(0x01020304);
     CNode peer_out{/*id=*/0,
-                   /*sock=*/nullptr,
                    /*addrIn=*/CAddress{CService{peer_out_in_addr, 8333}, NODE_NETWORK},
                    /*nKeyedNetGroupIn=*/0,
                    /*nLocalHostNonceIn=*/0,
@@ -681,7 +675,6 @@ BOOST_AUTO_TEST_CASE(get_local_addr_for_peer_port)
     in_addr peer_in_in_addr;
     peer_in_in_addr.s_addr = htonl(0x05060708);
     CNode peer_in{/*id=*/0,
-                  /*sock=*/nullptr,
                   /*addrIn=*/CAddress{CService{peer_in_in_addr, 8333}, NODE_NETWORK},
                   /*nKeyedNetGroupIn=*/0,
                   /*nLocalHostNonceIn=*/0,
@@ -818,7 +811,6 @@ BOOST_AUTO_TEST_CASE(initial_advertise_from_version_message)
     in_addr peer_in_addr;
     peer_in_addr.s_addr = htonl(0x01020304);
     CNode peer{/*id=*/0,
-               /*sock=*/nullptr,
                /*addrIn=*/CAddress{CService{peer_in_addr, 8333}, NODE_NETWORK},
                /*nKeyedNetGroupIn=*/0,
                /*nLocalHostNonceIn=*/0,
@@ -893,7 +885,6 @@ BOOST_AUTO_TEST_CASE(advertise_local_address)
 {
     auto CreatePeer = [](const CAddress& addr) {
         return std::make_unique<CNode>(/*id=*/0,
-                                       /*sock=*/nullptr,
                                        addr,
                                        /*nKeyedNetGroupIn=*/0,
                                        /*nLocalHostNonceIn=*/0,

src/test/util/net.h

@@ -15,6 +15,7 @@
 #include <node/eviction.h>
 #include <span.h>
 #include <sync.h>
+#include <util/check.h>
 #include <util/sock.h>
 
 #include <algorithm>
@@ -45,16 +46,23 @@ struct ConnmanTestMsg : public CConnman {
         m_peer_connect_timeout = timeout;
     }
 
-    std::vector<CNode*> TestNodes()
+    auto TestNodes()
     {
         LOCK(m_nodes_mutex);
         return m_nodes;
     }
 
+    void AddTestNode(CNode& node, std::unique_ptr<Sock>&& sock)
+    {
+        TestOnlyAddExistentConnection(node.GetId(), std::move(sock));
+        AddTestNode(node);
+    }
+
     void AddTestNode(CNode& node)
     {
         LOCK(m_nodes_mutex);
-        m_nodes.push_back(&node);
+        auto [_, inserted] = m_nodes.emplace(node.GetId(), &node);
+        Assert(inserted);
 
         if (node.IsManualOrFullOutboundConn()) ++m_network_conn_counts[node.addr.GetNetwork()];
     }
@@ -62,7 +70,7 @@ struct ConnmanTestMsg : public CConnman {
     void ClearTestNodes()
     {
         LOCK(m_nodes_mutex);
-        for (CNode* node : m_nodes) {
+        for (auto& [_, node] : m_nodes) {
             delete node;
         }
         m_nodes.clear();
@@ -88,8 +96,9 @@ struct ConnmanTestMsg : public CConnman {
 
     bool AlreadyConnectedPublic(const CAddress& addr) { return AlreadyConnectedToAddress(addr); };
 
-    CNode* ConnectNodePublic(PeerManager& peerman, const char* pszDest, ConnectionType conn_type)
-        EXCLUSIVE_LOCKS_REQUIRED(!m_unused_i2p_sessions_mutex);
+    CNode* ConnectNodePublic(PeerManager& peerman, const char* pszDest, ConnectionType conn_type);
+
+    using CConnman::MarkAsDisconnectAndCloseConnection;
 };
 
 constexpr ServiceFlags ALL_SERVICE_FLAGS[]{

src/test/util_string_tests.cpp

@@ -2,6 +2,7 @@
 // Distributed under the MIT software license, see the accompanying
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
+#include <util/strencodings.h>
 #include <util/string.h>
 
 #include <boost/test/unit_test.hpp>
@@ -146,4 +147,57 @@ BOOST_AUTO_TEST_CASE(ConstevalFormatString_NumSpec)
         HasReason{"tinyformat: Too many conversion specifiers in format string"});
 }
 
+BOOST_AUTO_TEST_CASE(case_insensitive_comparator_test)
+{
+    CaseInsensitiveComparator cmp;
+    BOOST_CHECK(cmp("A", "B"));
+    BOOST_CHECK(cmp("A", "b"));
+    BOOST_CHECK(cmp("a", "B"));
+    BOOST_CHECK(!cmp("B", "A"));
+    BOOST_CHECK(!cmp("B", "a"));
+    BOOST_CHECK(!cmp("b", "A"));
+}
+
+BOOST_AUTO_TEST_CASE(line_reader_test)
+{
+    {
+        // Check three lines terminated by \n, \r\n, and end of buffer, trimming whitespace
+        const std::vector<std::byte> input{StringToBuffer("once upon a time\n there was a dog \r\nwho liked food")};
+        LineReader reader(input, /*max_read=*/128);
+        std::optional<std::string> line1{reader.ReadLine()};
+        BOOST_CHECK_EQUAL(reader.Left(), 33);
+        std::optional<std::string> line2{reader.ReadLine()};
+        BOOST_CHECK_EQUAL(reader.Left(), 14);
+        std::optional<std::string> line3{reader.ReadLine()};
+        std::optional<std::string> line4{reader.ReadLine()};
+        BOOST_CHECK(line1);
+        BOOST_CHECK(line2);
+        BOOST_CHECK(line3);
+        BOOST_CHECK(!line4);
+        BOOST_CHECK_EQUAL(line1.value(), "once upon a time");
+        BOOST_CHECK_EQUAL(line2.value(), "there was a dog");
+        BOOST_CHECK_EQUAL(line3.value(), "who liked food");
+    }
+    {
+        // Do not exceed max_read while searching for EOL
+        const std::vector<std::byte> input1{StringToBuffer("once upon a time there was a dog\nwho liked food")};
+        LineReader reader1(input1, /*max_read=*/10);
+        BOOST_CHECK_THROW(reader1.ReadLine(), std::runtime_error);
+
+        const std::vector<std::byte> input2{StringToBuffer("once upon\n a time there was a dog who liked food")};
+        LineReader reader2(input2, /*max_read=*/10);
+        BOOST_CHECK_EQUAL(reader2.ReadLine(), "once upon");
+        BOOST_CHECK_THROW(reader2.ReadLine(), std::runtime_error);
+    }
+    {
+        // Read specific number of bytes regardless of max_read or \n unless buffer is too short
+        const std::vector<std::byte> input{StringToBuffer("once upon a time\n there was a dog \r\nwho liked food")};
+        LineReader reader(input, /*max_read=*/1);
+        BOOST_CHECK_EQUAL(reader.ReadLength(3), "onc");
+        BOOST_CHECK_EQUAL(reader.ReadLength(8), "e upon a");
+        BOOST_CHECK_EQUAL(reader.ReadLength(8), " time\n t");
+        BOOST_CHECK_THROW(reader.ReadLength(128), std::runtime_error);
+    }
+}
+
 BOOST_AUTO_TEST_SUITE_END()

src/test/util_tests.cpp

@@ -385,6 +385,13 @@ BOOST_AUTO_TEST_CASE(util_FormatISO8601Date)
     BOOST_CHECK_EQUAL(FormatISO8601Date(1317425777), "2011-09-30");
 }
 
+BOOST_AUTO_TEST_CASE(util_FormatRFC7231DateTime)
+{
+    BOOST_CHECK_EQUAL(FormatRFC7231DateTime(253402214400), "Fri, 31 Dec 9999 00:00:00 GMT");
+    BOOST_CHECK_EQUAL(FormatRFC7231DateTime(1717429609), "Mon, 03 Jun 2024 15:46:49 GMT");
+    BOOST_CHECK_EQUAL(FormatRFC7231DateTime(0), "Thu, 01 Jan 1970 00:00:00 GMT");
+}
+
 BOOST_AUTO_TEST_CASE(util_FormatMoney)
 {
     BOOST_CHECK_EQUAL(FormatMoney(0), "0.00");
@@ -1045,6 +1052,26 @@ BOOST_AUTO_TEST_CASE(test_ParseUInt64)
     BOOST_CHECK(!ParseUInt64("-1234", &n));
 }
 
+BOOST_AUTO_TEST_CASE(test_ParseUInt64Hex)
+{
+    uint64_t n;
+    // Valid values
+    BOOST_CHECK(ParseUInt64Hex("1234", nullptr));
+    BOOST_CHECK(ParseUInt64Hex("1234", &n) && n == 4660);
+    BOOST_CHECK(ParseUInt64Hex("a", &n) && n == 10);
+    BOOST_CHECK(ParseUInt64Hex("0000000a", &n) && n == 10);
+    BOOST_CHECK(ParseUInt64Hex("100", &n) && n == 256);
+    BOOST_CHECK(ParseUInt64Hex("DEADbeef", &n) && n == 3735928559);
+    BOOST_CHECK(ParseUInt64Hex("FfFfFfFf", &n) && n == 4294967295);
+    // Invalid values
+    BOOST_CHECK(!ParseUInt64Hex("123456789", &n));
+    BOOST_CHECK(!ParseUInt64Hex("", &n));
+    BOOST_CHECK(!ParseUInt64Hex("-1", &n));
+    BOOST_CHECK(!ParseUInt64Hex("10 00", &n));
+    BOOST_CHECK(!ParseUInt64Hex("1 ", &n));
+    BOOST_CHECK(!ParseUInt64Hex("0xAB", &n));
+}
+
 BOOST_AUTO_TEST_CASE(test_FormatParagraph)
 {
     BOOST_CHECK_EQUAL(FormatParagraph("", 79, 0), "");

src/util/strencodings.cpp

@@ -251,6 +251,24 @@ bool ParseUInt64(std::string_view str, uint64_t* out)
     return ParseIntegral<uint64_t>(str, out);
 }
 
+bool ParseUInt64Hex(std::string_view str, uint64_t* out)
+{
+    if (str.size() > 8) return false;
+    if (str.size() < 1) return false;
+    uint64_t total{0};
+    auto it = str.begin();
+    while (it != str.end()) {
+        auto v = HexDigit(*(it++));
+        if (v < 0) return false;
+        total <<= 4;
+        total |= v;
+    }
+    if (out != nullptr) {
+        *out = total;
+    }
+    return true;
+}
+
 std::string FormatParagraph(std::string_view in, size_t width, size_t indent)
 {
     assert(width >= indent);
@@ -479,3 +497,10 @@ std::optional<uint64_t> ParseByteUnits(std::string_view str, ByteUnit default_mu
     }
     return *parsed_num * unit_amount;
 }
+
+std::vector<std::byte> StringToBuffer(const std::string& str)
+{
+    return std::vector<std::byte>(
+        reinterpret_cast<const std::byte*>(str.data()),
+        reinterpret_cast<const std::byte*>(str.data() + str.size()));
+}

src/util/strencodings.h

@@ -229,6 +229,14 @@ std::optional<T> ToIntegral(std::string_view str)
  */
 [[nodiscard]] bool ParseUInt64(std::string_view str, uint64_t *out);
 
+/**
+ * Convert hexadecimal string to unsigned 64-bit integer, with 4-bit
+ * resolution (odd length strings are acceptable without leading "0")
+ * @returns true if the entire string could be parsed as valid integer,
+ *   false if not, or in case of overflow.
+ */
+[[nodiscard]] bool ParseUInt64Hex(std::string_view str, uint64_t *out);
+
 /**
  * Format a paragraph of text to a fixed width, adding spaces for
  * indentation to any added line.
@@ -367,6 +375,15 @@ std::string Capitalize(std::string str);
  */
 std::optional<uint64_t> ParseByteUnits(std::string_view str, ByteUnit default_multiplier);
 
+/**
+ * Returns a byte vector filled with data from a string. Used to test string-
+ * encoded data from a socket like HTTP headers.
+ *
+ * @param[in] str                  the string to convert into bytes
+ * @returns                        byte vector
+ */
+std::vector<std::byte> StringToBuffer(const std::string& str);
+
 namespace util {
 /** consteval version of HexDigit() without the lookup table. */
 consteval uint8_t ConstevalHexDigit(const char c)
@@ -395,6 +412,24 @@ struct Hex {
 };
 } // namespace detail
 
+struct CaseInsensitiveComparator {
+    // Helper function for locale-independent tolower
+    static char tolowercase(char c)
+    {
+        return static_cast<char>(ToLower(static_cast<unsigned char>(c)));
+    }
+
+    bool operator()(const std::string& s1, const std::string& s2) const
+    {
+        return std::lexicographical_compare(
+            s1.begin(), s1.end(),
+            s2.begin(), s2.end(),
+            [](unsigned char c1, unsigned char c2) {
+                return tolowercase(c1) < tolowercase(c2);
+            });
+    }
+};
+
 /**
  * ""_hex is a compile-time user-defined literal returning a
  * `std::array<std::byte>`, equivalent to ParseHex(). Variants provided:

src/util/string.cpp

@@ -13,4 +13,41 @@ void ReplaceAll(std::string& in_out, const std::string& search, const std::strin
     if (search.empty()) return;
     in_out = std::regex_replace(in_out, std::regex(search), substitute);
 }
+
+LineReader::LineReader(std::span<const std::byte> buffer, size_t max_read)
+    : start(buffer.begin()), end(buffer.end()), max_read(max_read), it(buffer.begin()) {}
+
+std::optional<std::string> LineReader::ReadLine()
+{
+    if (it == end) {
+        return std::nullopt;
+    }
+
+    auto line_start = it;
+    std::string line{};
+    while (it != end) {
+        char c = static_cast<char>(*it);
+        line += c;
+        ++it;
+        if (c == '\n') break;
+        if ((size_t)std::distance(line_start, it) >= max_read) throw std::runtime_error("max_read exceeded by LineReader");
+    }
+
+    line = TrimString(line); // delete trailing \r and/or \n
+    return line;
+}
+
+// Ignores max_read but won't overflow
+std::string LineReader::ReadLength(size_t len)
+{
+    if (Left() < len) throw std::runtime_error("Not enough data in buffer");
+    std::string out(reinterpret_cast<const char*>(&(*it)), len);
+    it += len;
+    return out;
+}
+
+size_t LineReader::Left() const
+{
+    return std::distance(it, end);
+}
 } // namespace util

src/util/string.h

@@ -11,6 +11,7 @@
 #include <cstdint>
 #include <cstring>
 #include <locale>
+#include <optional>
 #include <sstream>
 #include <string>      // IWYU pragma: export
 #include <string_view> // IWYU pragma: export
@@ -248,6 +249,25 @@ template <typename T1, size_t PREFIX_LEN>
     return obj.size() >= PREFIX_LEN &&
            std::equal(std::begin(prefix), std::end(prefix), std::begin(obj));
 }
+
+struct LineReader {
+    const std::span<const std::byte>::iterator start;
+    const std::span<const std::byte>::iterator end;
+    const size_t max_read;
+    std::span<const std::byte>::iterator it;
+
+    explicit LineReader(std::span<const std::byte> buffer, size_t max_read);
+
+    // Returns a string from current iterator position up to next \n
+    // and advances iterator, does not return trailing \n or \r.
+    // Will not search for \n past max_read.
+    std::optional<std::string> ReadLine();
+    // Returns string from current iterator position of specified length
+    // and advances iterator. May exceed max_read but will not read past end of buffer.
+    std::string ReadLength(size_t len);
+    // Returns remaining size of bytes in buffer
+    size_t Left() const;
+};
 } // namespace util
 
 #endif // BITCOIN_UTIL_STRING_H

src/util/time.cpp

@@ -17,6 +17,9 @@
 #include <string_view>
 #include <thread>
 
+static const std::string weekdays[7] = {"Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"};
+static const std::string months[12] = {"Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"};
+
 void UninterruptibleSleep(const std::chrono::microseconds& n) { std::this_thread::sleep_for(n); }
 
 static std::atomic<std::chrono::seconds> g_mock_time{}; //!< For testing
@@ -116,6 +119,20 @@ std::optional<int64_t> ParseISO8601DateTime(std::string_view str)
     return int64_t{TicksSinceEpoch<std::chrono::seconds>(tp)};
 }
 
+std::string FormatRFC7231DateTime(int64_t nTime)
+{
+    const std::chrono::sys_seconds secs{std::chrono::seconds{nTime}};
+    const auto days{std::chrono::floor<std::chrono::days>(secs)};
+    // 1970-01-01 was a Thursday
+    std::string weekday{weekdays[(days.time_since_epoch().count() + 4) % 7]};
+    const std::chrono::year_month_day ymd{days};
+    std::string month{months[unsigned{ymd.month()} - 1]};
+    const std::chrono::hh_mm_ss hms{secs - days};
+    // examples: Mon, 27 Jul 2009 12:28:53 GMT
+    //           Fri, 31 May 2024 19:18:04 GMT
+    return strprintf("%03s, %02u %03s %04i %02i:%02i:%02i GMT", weekday, unsigned{ymd.day()}, month, signed{ymd.year()}, hms.hours().count(), hms.minutes().count(), hms.seconds().count());
+}
+
 struct timeval MillisToTimeval(int64_t nTimeout)
 {
     struct timeval timeout;

src/util/time.h

@@ -134,6 +134,12 @@ std::string FormatISO8601DateTime(int64_t nTime);
 std::string FormatISO8601Date(int64_t nTime);
 std::optional<int64_t> ParseISO8601DateTime(std::string_view str);
 
+/**
+ * RFC7231 formatting https://www.rfc-editor.org/rfc/rfc7231#section-7.1.1.1
+ * Used in HTTP/1.1 responses
+ */
+std::string FormatRFC7231DateTime(int64_t nTime);
+
 /**
  * Convert milliseconds to a struct timeval for e.g. select.
  */

test/functional/feature_init.py

@@ -88,7 +88,7 @@ class InitTest(BitcoinTestFramework):
 
         args = ['-txindex=1', '-blockfilterindex=1', '-coinstatsindex=1']
         for terminate_line in lines_to_terminate_after:
-            self.log.info(f"Starting node and will exit after line {terminate_line}")
+            self.log.info(f"Starting node and will terminate after line {terminate_line}")
             with node.busy_wait_for_debug_log([terminate_line]):
                 if platform.system() == 'Windows':
                     # CREATE_NEW_PROCESS_GROUP is required in order to be able
@@ -108,12 +108,22 @@ class InitTest(BitcoinTestFramework):
             'blocks/index/*.ldb': 'Error opening block database.',
             'chainstate/*.ldb': 'Error opening coins database.',
             'blocks/blk*.dat': 'Error loading block database.',
+            'indexes/txindex/MANIFEST*': 'LevelDB error: Corruption: CURRENT points to a non-existent file',
+            # Removing these files does not result in a startup error:
+            # 'indexes/blockfilter/basic/*.dat', 'indexes/blockfilter/basic/db/*.*', 'indexes/coinstats/db/*.*',
+            # 'indexes/txindex/*.log', 'indexes/txindex/CURRENT', 'indexes/txindex/LOCK'
         }
 
         files_to_perturb = {
             'blocks/index/*.ldb': 'Error loading block database.',
             'chainstate/*.ldb': 'Error opening coins database.',
             'blocks/blk*.dat': 'Corrupted block database detected.',
+            'indexes/blockfilter/basic/db/*.*': 'LevelDB error: Corruption',
+            'indexes/coinstats/db/*.*': 'LevelDB error: Corruption',
+            'indexes/txindex/*.log': 'LevelDB error: Corruption',
+            'indexes/txindex/CURRENT': 'LevelDB error: Corruption',
+            # Perturbing these files does not result in a startup error:
+            # 'indexes/blockfilter/basic/*.dat', 'indexes/txindex/MANIFEST*', 'indexes/txindex/LOCK'
         }
 
         for file_patt, err_fragment in files_to_delete.items():
@@ -135,9 +145,10 @@ class InitTest(BitcoinTestFramework):
             self.stop_node(0)
 
         self.log.info("Test startup errors after perturbing certain essential files")
+        dirs = ["blocks", "chainstate", "indexes"]
         for file_patt, err_fragment in files_to_perturb.items():
-            shutil.copytree(node.chain_path / "blocks", node.chain_path / "blocks_bak")
-            shutil.copytree(node.chain_path / "chainstate", node.chain_path / "chainstate_bak")
+            for dir in dirs:
+                shutil.copytree(node.chain_path / dir, node.chain_path / f"{dir}_bak")
             target_files = list(node.chain_path.glob(file_patt))
 
             for target_file in target_files:
@@ -151,10 +162,9 @@ class InitTest(BitcoinTestFramework):
 
             start_expecting_error(err_fragment)
 
-            shutil.rmtree(node.chain_path / "blocks")
-            shutil.rmtree(node.chain_path / "chainstate")
-            shutil.move(node.chain_path / "blocks_bak", node.chain_path / "blocks")
-            shutil.move(node.chain_path / "chainstate_bak", node.chain_path / "chainstate")
+            for dir in dirs:
+                shutil.rmtree(node.chain_path / dir)
+                shutil.move(node.chain_path / f"{dir}_bak", node.chain_path / dir)
 
     def init_pid_test(self):
         BITCOIN_PID_FILENAME_CUSTOM = "my_fancy_bitcoin_pid_file.foobar"

test/functional/feature_port.py

@@ -29,23 +29,23 @@ class PortTest(BitcoinTestFramework):
         port2 = p2p_port(self.num_nodes + 5)
 
         self.log.info("When starting with -port, bitcoind binds to it and uses port + 1 for an onion bind")
-        with node.assert_debug_log(expected_msgs=[f'Bound to 0.0.0.0:{port1}', f'Bound to 127.0.0.1:{port1 + 1}']):
+        with node.assert_debug_log(expected_msgs=[f'Bound to and listening on 0.0.0.0:{port1}', f'Bound to and listening on 127.0.0.1:{port1 + 1}']):
             self.restart_node(0, extra_args=["-listen", f"-port={port1}"])
 
         self.log.info("When specifying -port multiple times, only the last one is taken")
-        with node.assert_debug_log(expected_msgs=[f'Bound to 0.0.0.0:{port2}', f'Bound to 127.0.0.1:{port2 + 1}'], unexpected_msgs=[f'Bound to 0.0.0.0:{port1}']):
+        with node.assert_debug_log(expected_msgs=[f'Bound to and listening on 0.0.0.0:{port2}', f'Bound to and listening on 127.0.0.1:{port2 + 1}'], unexpected_msgs=[f'Bound to and listening on 0.0.0.0:{port1}']):
             self.restart_node(0, extra_args=["-listen", f"-port={port1}", f"-port={port2}"])
 
         self.log.info("When specifying ports with both -port and -bind, the one from -port is ignored")
-        with node.assert_debug_log(expected_msgs=[f'Bound to 0.0.0.0:{port2}'], unexpected_msgs=[f'Bound to 0.0.0.0:{port1}']):
+        with node.assert_debug_log(expected_msgs=[f'Bound to and listening on 0.0.0.0:{port2}'], unexpected_msgs=[f'Bound to and listening on 0.0.0.0:{port1}']):
             self.restart_node(0, extra_args=["-listen", f"-port={port1}", f"-bind=0.0.0.0:{port2}"])
 
         self.log.info("When -bind specifies no port, the values from -port and -bind are combined")
-        with self.nodes[0].assert_debug_log(expected_msgs=[f'Bound to 0.0.0.0:{port1}']):
+        with self.nodes[0].assert_debug_log(expected_msgs=[f'Bound to and listening on 0.0.0.0:{port1}']):
             self.restart_node(0, extra_args=["-listen", f"-port={port1}", "-bind=0.0.0.0"])
 
         self.log.info("When an onion bind specifies no port, the value from -port, incremented by 1, is taken")
-        with self.nodes[0].assert_debug_log(expected_msgs=[f'Bound to 127.0.0.1:{port1 + 1}']):
+        with self.nodes[0].assert_debug_log(expected_msgs=[f'Bound to and listening on 127.0.0.1:{port1 + 1}']):
             self.restart_node(0, extra_args=["-listen", f"-port={port1}", "-bind=127.0.0.1=onion"])
 
         self.log.info("Invalid values for -port raise errors")

test/functional/interface_http.py

@@ -8,6 +8,7 @@ from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import assert_equal, str_to_b64str
 
 import http.client
+import time
 import urllib.parse
 
 class HTTPBasicsTest (BitcoinTestFramework):
@@ -104,6 +105,74 @@ class HTTPBasicsTest (BitcoinTestFramework):
         out1 = conn.getresponse()
         assert_equal(out1.status, http.client.BAD_REQUEST)
 
+        self.log.info("Check HTTP request encoded with chunked transfer")
+        headers_chunked = headers.copy()
+        headers_chunked.update({"Transfer-encoding": "chunked"})
+        body_chunked = [
+            b'{"method": "submitblock", "params": ["',
+            b'0A' * 1000000,
+            b'0B' * 1000000,
+            b'0C' * 1000000,
+            b'0D' * 1000000,
+            b'"]}'
+        ]
+        conn = http.client.HTTPConnection(urlNode2.hostname, urlNode2.port)
+        conn.connect()
+        conn.request(
+            method='POST',
+            url='/',
+            body=iter(body_chunked),
+            headers=headers_chunked,
+            encode_chunked=True)
+        out1 = conn.getresponse().read()
+        assert out1 == b'{"result":"high-hash","error":null}\n'
+
+        self.log.info("Check -rpcservertimeout")
+        self.restart_node(2, extra_args=["-rpcservertimeout=1"])
+        # This is the amount of time the server will wait for a client to
+        # send a complete request. Test it by sending an incomplete but
+        # so-far otherwise well-formed HTTP request, and never finishing it.
+
+        # Copied from http_incomplete_test_() in regress_http.c in libevent.
+        # A complete request would have an additional "\r\n" at the end.
+        http_request = "GET /test1 HTTP/1.1\r\nHost: somehost\r\n"
+
+        # Get the underlying socket from HTTP connection so we can send something unusual
+        conn = http.client.HTTPConnection(urlNode2.hostname, urlNode2.port)
+        conn.connect()
+        sock = conn.sock
+        sock.sendall(http_request.encode("utf-8"))
+        # Wait for response, but expect a timeout disconnection after 1 second
+        start = time.time()
+        res = sock.recv(1024)
+        stop = time.time()
+        assert res == b""
+        assert stop - start >= 1
+        # definitely closed
+        try:
+            conn.request('GET', '/')
+            conn.getresponse()
+        except ConnectionResetError:
+            pass
+
+        # Sanity check
+        http_request = "GET /test2 HTTP/1.1\r\nHost: somehost\r\n\r\n"
+        conn = http.client.HTTPConnection(urlNode2.hostname, urlNode2.port)
+        conn.connect()
+        sock = conn.sock
+        sock.sendall(http_request.encode("utf-8"))
+        res = sock.recv(1024)
+        assert res.startswith(b"HTTP/1.1 404 Not Found")
+        # still open
+        conn.request('GET', '/')
+        conn.getresponse()
+
+        # Because we have set -rpcservertimeout so low, the persistent connection
+        # created by AuthServiceProxy for this node when the test framework
+        # started has likely closed. Force the test framework to use a fresh
+        # new connection for the next RPC otherwise the cleanup process
+        # calling `stop` will raise a connection error.
+        self.nodes[2]._set_conn()
 
 if __name__ == '__main__':
     HTTPBasicsTest(__file__).main()

test/functional/interface_rest.py

@@ -282,10 +282,12 @@ class RESTTest (BitcoinTestFramework):
         assert_equal(len(json_obj), 1)  # ensure that there is one header in the json response
         assert_equal(json_obj[0]['hash'], bb_hash)  # request/response hash should be the same
 
-        # Check invalid uri (% symbol at the end of the request)
-        for invalid_uri in [f"/headers/{bb_hash}%", f"/blockfilterheaders/basic/{bb_hash}%", "/mempool/contents.json?%"]:
+        # Check tolerance for invalid URI (% symbol at the end of the request)
+        for invalid_uri in [f"/headers/{bb_hash}%", f"/blockfilterheaders/basic/{bb_hash}%"]:
             resp = self.test_rest_request(invalid_uri, ret_type=RetType.OBJ, status=400)
-            assert_equal(resp.read().decode('utf-8').rstrip(), "URI parsing failed, it likely contained RFC 3986 invalid characters")
+            assert_equal(resp.read().decode('utf-8').rstrip(), f"Invalid hash: {bb_hash}%")
+        resp = self.test_rest_request("/mempool/contents.json?%", ret_type=RetType.OBJ, status=200)
+        assert_equal(resp.read().decode('utf-8').rstrip(), "{}")
 
         # Compare with normal RPC block response
         rpc_block_json = self.nodes[0].getblock(bb_hash)

test/functional/mempool_accept.py

@@ -45,6 +45,7 @@ from test_framework.util import (
     assert_equal,
     assert_greater_than,
     assert_raises_rpc_error,
+    sync_txindex,
 )
 from test_framework.wallet import MiniWallet
 from test_framework.wallet_util import generate_keypair
@@ -270,6 +271,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
 
         self.log.info('A coinbase transaction')
         # Pick the input of the first tx we created, so it has to be a coinbase tx
+        sync_txindex(self, node)
         raw_tx_coinbase_spent = node.getrawtransaction(txid=node.decoderawtransaction(hexstring=raw_tx_in_block)['vin'][0]['txid'])
         tx = tx_from_hex(raw_tx_coinbase_spent)
         self.check_mempool_result(

test/functional/rpc_rawtransaction.py

@@ -34,6 +34,7 @@ from test_framework.util import (
     assert_equal,
     assert_greater_than,
     assert_raises_rpc_error,
+    sync_txindex,
 )
 from test_framework.wallet import (
     getnewdestination,
@@ -70,7 +71,7 @@ class RawTransactionsTest(BitcoinTestFramework):
         self.num_nodes = 3
         self.extra_args = [
             ["-txindex"],
-            ["-txindex"],
+            [],
             ["-fastprune", "-prune=1"],
         ]
         # whitelist peers to speed up tx relay / mempool sync
@@ -109,6 +110,7 @@ class RawTransactionsTest(BitcoinTestFramework):
             self.log.info(f"Test getrawtransaction {'with' if n == 0 else 'without'} -txindex")
 
             if n == 0:
+                sync_txindex(self, self.nodes[n])
                 # With -txindex.
                 # 1. valid parameters - only supply txid
                 assert_equal(self.nodes[n].getrawtransaction(txId), tx['hex'])

test/functional/rpc_txoutproof.py

@@ -12,6 +12,7 @@ from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
     assert_raises_rpc_error,
+    sync_txindex,
 )
 from test_framework.wallet import MiniWallet
 
@@ -77,6 +78,7 @@ class MerkleBlockTest(BitcoinTestFramework):
         assert_equal(sorted(self.nodes[0].verifytxoutproof(self.nodes[0].gettxoutproof([txid1, txid2]))), sorted(txlist))
         assert_equal(sorted(self.nodes[0].verifytxoutproof(self.nodes[0].gettxoutproof([txid2, txid1]))), sorted(txlist))
         # We can always get a proof if we have a -txindex
+        sync_txindex(self, self.nodes[1])
         assert_equal(self.nodes[0].verifytxoutproof(self.nodes[1].gettxoutproof([txid_spent])), [txid_spent])
         # We can't get a proof if we specify transactions from different blocks
         assert_raises_rpc_error(-5, "Not all transactions found in specified or retrieved block", self.nodes[0].gettxoutproof, [txid1, txid3])

test/functional/test_framework/util.py

@@ -592,3 +592,10 @@ def find_vout_for_address(node, txid, addr):
         if addr == tx["vout"][i]["scriptPubKey"]["address"]:
             return i
     raise RuntimeError("Vout not found for address: txid=%s, addr=%s" % (txid, addr))
+
+
+def sync_txindex(test_framework, node):
+    test_framework.log.debug("Waiting for node txindex to sync")
+    sync_start = int(time.time())
+    test_framework.wait_until(lambda: node.getindexinfo("txindex")["txindex"]["synced"])
+    test_framework.log.debug(f"Synced in {time.time() - sync_start} seconds")

test/functional/wallet_avoid_mixing_output_types.py

@@ -117,7 +117,6 @@ class AddressInputTypeGrouping(BitcoinTestFramework):
         self.extra_args = [
             [
                 "-addresstype=bech32",
-                "-txindex",
             ],
             [
                 "-addresstype=p2sh-segwit",