ipc, refactor: Add Stream type alias and use it

diff --git a/src/ipc/capnp/protocol.cpp b/src/ipc/capnp/protocol.cpp
index 0790986d..64b1e9a9 100644
--- a/src/ipc/capnp/protocol.cpp
+++ b/src/ipc/capnp/protocol.cpp
@@ -78,10 +78,10 @@ public:
         if (m_loop_thread.joinable()) m_loop_thread.join();
         assert(!m_loop);
     };
-    std::unique_ptr<interfaces::Init> connect(mp::SocketId socket) override
+    std::unique_ptr<interfaces::Init> connect(mp::Stream stream) override
     {
         startLoop();
-        return mp::ConnectStream<messages::Init>(*m_loop, socket);
+        return mp::ConnectStream<messages::Init>(*m_loop, std::move(stream));
     }
     void listen(mp::SocketId listen_fd, interfaces::Init& init) override
     {
@@ -91,7 +91,7 @@ public:
         }
         mp::ListenConnections<messages::Init>(*m_loop, listen_fd, init);
     }
-    void serve(mp::SocketId socket, interfaces::Init& init, const std::function<void()>& ready_fn = {}) override
+    void serve(interfaces::Init& init, const std::function<mp::Stream()>& make_stream) override
     {
         assert(!m_loop);
         mp::g_thread_context.thread_name = mp::ThreadName(m_exe_name);
@@ -100,8 +100,7 @@ public:
             .log_level = GetRequestedIPCLogLevel()
         };
         m_loop.emplace(m_exe_name, std::move(opts), &m_context);
-        if (ready_fn) ready_fn();
-        mp::ServeStream<messages::Init>(*m_loop, socket, init);
+        mp::ServeStream<messages::Init>(*m_loop, make_stream(), init);
         m_parent_connection = &m_loop->m_incoming_connections.back();
         m_loop->loop();
         m_loop.reset();
@@ -116,6 +115,11 @@ public:
             m_loop->m_incoming_connections.remove_if([this](mp::Connection& c) { return &c != m_parent_connection; });
         });
     }
+    mp::Stream makeStream(mp::SocketId socket) override
+    {
+        startLoop();
+        return mp::MakeStream(*m_loop, socket);
+    }
     void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
     {
         mp::ProxyTypeRegister::types().at(type)(iface).cleanup_fns.emplace_back(std::move(cleanup));
diff --git a/src/ipc/interfaces.cpp b/src/ipc/interfaces.cpp
index 66b5e8ec..75a854c2 100644
--- a/src/ipc/interfaces.cpp
+++ b/src/ipc/interfaces.cpp
@@ -65,7 +65,7 @@ public:
         mp::ProcessId pid;
         mp::SocketId fd = m_process->spawn(new_exe_name, m_process_argv0, pid);
         LogDebug(::BCLog::IPC, "Process %s pid %i launched\n", new_exe_name, pid);
-        auto init = m_protocol->connect(fd);
+        auto init = m_protocol->connect(m_protocol->makeStream(fd));
         Ipc::addCleanup(*init, [this, new_exe_name, pid] {
             int status = m_process->waitSpawned(pid);
             LogDebug(::BCLog::IPC, "Process %s pid %i exited with status %i\n", new_exe_name, pid, status);
@@ -80,7 +80,7 @@ public:
             return false;
         }
         IgnoreCtrlC(strprintf("[%s] SIGINT received — waiting for parent to shut down.\n", m_exe_name));
-        m_protocol->serve(socket, m_init);
+        m_protocol->serve(m_init, [&] { return m_protocol->makeStream(socket); } );
         exit_status = EXIT_SUCCESS;
         return true;
     }
@@ -109,7 +109,7 @@ public:
         } else {
             fd = m_process->connect(gArgs.GetDataDirNet(), "bitcoin-node", address);
         }
-        return m_protocol->connect(fd);
+        return m_protocol->connect(m_protocol->makeStream(fd));
     }
     void listenAddress(std::string& address) override
     {
diff --git a/src/ipc/protocol.h b/src/ipc/protocol.h
index 4c3c1bdc..66aab9fb 100644
--- a/src/ipc/protocol.h
+++ b/src/ipc/protocol.h
@@ -24,8 +24,8 @@ class Protocol
 public:
     virtual ~Protocol() = default;
 
-    //! Return Init interface that forwards requests over given socket descriptor.
-    //! Socket communication is handled on a background thread.
+    //! Return Init interface that forwards requests over given connection
+    //! stream. Socket communication is handled on a background thread.
     //!
     //! @note It could be potentially useful in the future to add
     //! std::function<void()> on_disconnect callback argument here. But there
@@ -33,31 +33,31 @@ public:
     //! up its own state (calling ProxyServer destructors, etc) on disconnect,
     //! and any client calls will just throw ipc::Exception errors after a
     //! disconnect.
-    virtual std::unique_ptr<interfaces::Init> connect(mp::SocketId fd) = 0;
+    virtual std::unique_ptr<interfaces::Init> connect(mp::Stream stream) = 0;
 
     //! Listen for connections on provided socket id, accept them, and handle
     //! requests on accepted connections. This method doesn't block, and
     //! performs I/O on a background thread.
     virtual void listen(mp::SocketId listen_fd, interfaces::Init& init) = 0;
 
-    //! Handle requests on provided socket descriptor, forwarding them to the
-    //! provided Init interface. Socket communication is handled on the
-    //! current thread, and this call blocks until the socket is closed.
+    //! Handle requests from a stream provided by the make_stream callback,
+    //! forwarding them to the provided Init interface. Socket communication is
+    //! handled on the current thread, and this call blocks until the socket is
+    //! closed. A callback is used to specify the stream because this method
+    //! initializes the event loop and it may not be possible to create the
+    //! stream before the event loop is initialized.
     //!
-    //! @note: If this method is called, it needs be called before connect() or
-    //! listen() methods, because for ease of implementation it's inflexible and
-    //! always runs the event loop in the foreground thread. It can share its
-    //! event loop with the other methods but can't share an event loop that was
-    //! created by them. This isn't really a problem because serve() is only
-    //! called by spawned child processes that call it immediately to
+    //! @note: If this method is called, it needs to be called before connect()
+    //! or listen() methods, because for ease of implementation this method is
+    //! inflexible and always runs the event loop in the foreground thread. It
+    //! can share its event loop with the other methods but can't share an event
+    //! loop that was created by them. This isn't a problem because serve() is
+    //! only called by spawned child processes that call it immediately to
     //! communicate back with parent processes.
-    //
-    //! The optional `ready_fn` callback will be called after the event loop is
-    //! created but before it is started. This can be useful in tests to trigger
-    //! client connections from another thread as soon as the event loop is
-    //! available, but should not be necessary in normal code which starts
-    //! clients and servers independently.
-    virtual void serve(mp::SocketId fd, interfaces::Init& init, const std::function<void()>& ready_fn = {}) = 0;
+    virtual void serve(interfaces::Init& init, const std::function<mp::Stream()>& make_stream) = 0;
+
+    //! Make stream object from socket id.
+    virtual mp::Stream makeStream(mp::SocketId socket) = 0;
 
     //! Disconnect any incoming connections that are still connected.
     virtual void disconnectIncoming() = 0;
diff --git a/src/ipc/test/ipc_tests.cpp b/src/ipc/test/ipc_tests.cpp
index f09c2d16..e353a7ee 100644
--- a/src/ipc/test/ipc_tests.cpp
+++ b/src/ipc/test/ipc_tests.cpp
@@ -129,16 +129,20 @@ void IpcPipeTest()
 //! Test ipc::Protocol connect() and serve() methods connecting over a socketpair.
 void IpcSocketPairTest()
 {
-    mp::SocketId fds[2];
-    BOOST_CHECK_EQUAL(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
     std::unique_ptr<interfaces::Init> init{std::make_unique<TestInit>()};
     std::unique_ptr<ipc::Protocol> protocol{ipc::capnp::MakeCapnpProtocol("IpcSocketPairTest")};
+    mp::Stream client_stream;
     std::promise<void> promise;
     std::thread thread([&]() {
-        protocol->serve(fds[0], *init, [&] { promise.set_value(); });
+        protocol->serve(*init, [&] {
+            auto pair{mp::SocketPair()};
+            client_stream = protocol->makeStream(pair[0]);
+            promise.set_value();
+            return protocol->makeStream(pair[1]);
+        });
     });
     promise.get_future().wait();
-    std::unique_ptr<interfaces::Init> remote_init{protocol->connect(fds[1])};
+    std::unique_ptr<interfaces::Init> remote_init{protocol->connect(std::move(client_stream))};
     std::unique_ptr<interfaces::Echo> remote_echo{remote_init->makeEcho()};
     BOOST_CHECK_EQUAL(remote_echo->echo("echo test"), "echo test");
     remote_echo.reset();
@@ -169,7 +173,7 @@ void IpcSocketTest(const fs::path& datadir)
         std::string address{connect_address};
         mp::SocketId connect_fd{process->connect(datadir, "test_bitcoin", address)};
         BOOST_CHECK_EQUAL(address, connect_address);
-        std::unique_ptr<interfaces::Init> remote_init{protocol->connect(connect_fd)};
+        std::unique_ptr<interfaces::Init> remote_init{protocol->connect(protocol->makeStream(connect_fd))};
         std::unique_ptr<interfaces::Echo> remote_echo{remote_init->makeEcho()};
         BOOST_CHECK_EQUAL(remote_echo->echo("echo test"), "echo test");
     }};
diff --git a/src/ipc/util.h b/src/ipc/util.h
index 5e591d7b..ec36b418 100644
--- a/src/ipc/util.h
+++ b/src/ipc/util.h
@@ -5,18 +5,35 @@
 #ifndef BITCOIN_IPC_UTIL_H
 #define BITCOIN_IPC_UTIL_H
 
+#include <array>
 #include <cstdint>
+#include <kj/debug.h>
 #include <mp/util.h>
 #include <mp/version.h>
+#include <sys/socket.h>
 
 namespace mp {
 // Definitions that can be deleted when libmultiprocess subtree is updated to
 // v14. Having these allows Bitcoin Core changes to be decoupled from
 // libmultiprocess changes so they don't have to be reviewed in a single PR.
 #if MP_MAJOR_VERSION < 14
+class EventLoop;
 using ProcessId = int;
 using SocketId = int;
 constexpr SocketId SocketError{-1};
+
+using Stream = SocketId;
+inline Stream MakeStream(EventLoop&, SocketId socket)
+{
+    return socket;
+}
+
+inline std::array<SocketId, 2> SocketPair()
+{
+    int pair[2];
+    KJ_SYSCALL(socketpair(AF_UNIX, SOCK_STREAM, 0, pair));
+    return {pair[0], pair[1]};
+}
 #endif
 } // namespace mp