core(socket): outbound write queue — stop overlapping composed writes losing shares (#874)

core::Socket::write started a composed boost::asio::async_write immediately,
with no outbound queue. Asio forbids initiating a second composed write on a
descriptor before the first completes: the two async_write_some continuations
are serviced FIFO per descriptor, so any message needing more than one round
gets bytes from the other message spliced into its middle. The peer sees a bad
length/checksum and drops us.

This is reachable on every coin. send_shares issues three back-to-back writes
on one socket in one handler turn (remember_tx -> shares -> forget_tx):
dash/node.cpp:1346/1357/1362, ltc:756/777/784, btc:757/778/785,
dgb:737/758/765, bch:760/781/788. handle_version writes getaddrs/addrme before
the up-to-32 KB have_tx advert. Because send_shares reports its hashes as
"sent" on SUBMISSION and broadcast_share never retries, a spliced frame loses
that share to that peer permanently — silent PPLNS loss with no counter and no
log line.

Fix: a per-socket std::deque of framed buffers plus a writing_ in-flight flag.
write() frames on the caller's thread exactly as before, queues, and starts the
drain only when nothing is in flight; do_write() starts the next composed write
only from the previous one's completion handler, so at most ONE composed
async_write is ever outstanding per socket. Same shape as the working
StratumSession::send_response/do_write already in core/stratum_server.cpp.

Thread-safety: writes are submitted from node/compute threads as well as the
io_context thread, which is how the bug is reachable. m_write_mutex is a LEAF
lock held only around the deque/flag mutation — never across async_write
initiation, an m_node callback, or any other lock — so no new lock-order edge.
The empty-queue check and the writing_ clear happen under the same lock, so a
concurrent submit can never lose the wakeup.

Semantics preserved: on write error the whole queued chain fails together and
error() is reported once, exactly as a single failed write did; a write onto an
already-closed socket stays a silent no-op; the single-write path is
byte-identical.

Tests (folded into the existing allowlisted core_test target, never a new
add_executable — that is the "Not Run" trap): src/core/test/
socket_write_queue_test.cpp drives a real loopback TCP pair with a squeezed
send buffer so each message needs many write_some rounds, then asserts
overlapping writes arrive concatenated in submission order, that concurrent
writers from several threads still produce whole checksum-valid frames, that a
single write is byte-identical, and that writes on a closed socket are dropped
without an error callback. The first two FAIL on the pre-fix tree (byte streams
diverge mid-payload with the next message's magic prefix spliced in).

Fixes #863

Author: frstrtr

src/core/socket.cpp

@@ -120,18 +120,92 @@ void Socket::write(std::unique_ptr<RawMessage> msg_data)
     }
 
     auto packet = std::make_shared<PackStream>(Packet::from_message(m_node->get_prefix(), msg_data));
+
+    // Queue, then start the drain ONLY if no composed write is in flight. The
+    // framing is done here (on the caller's thread, as before) so the bytes are
+    // captured in submission order; the wire order is then exactly the order in
+    // which callers reached this line. See the m_write_queue comment in
+    // socket.hpp for why overlapping composed writes cost shares.
+    bool start_drain = false;
+    {
+        std::lock_guard<std::mutex> lock(m_write_mutex);
+        m_write_queue.push_back(std::move(packet));
+        if (!m_writing)
+        {
+            m_writing = true;
+            start_drain = true;
+        }
+    }
+
+    // Outside the lock: do_write() initiates asio work and must never run with
+    // m_write_mutex held (leaf-lock discipline).
+    if (start_drain)
+        do_write();
+}
+
+void Socket::fail_write_queue()
+{
+    std::deque<std::shared_ptr<PackStream>> dropped;
+    {
+        std::lock_guard<std::mutex> lock(m_write_mutex);
+        dropped.swap(m_write_queue);
+        m_writing = false;
+    }
+    // `dropped` is released here, outside the lock.
+}
+
+void Socket::do_write()
+{
+    std::shared_ptr<PackStream> packet;
+    {
+        std::lock_guard<std::mutex> lock(m_write_mutex);
+        if (m_write_queue.empty())
+        {
+            m_writing = false;
+            return;
+        }
+        packet = m_write_queue.front();
+    }
+
+    // Socket closed under us between queueing and draining: fail the whole
+    // chain rather than leave buffers pinned behind a stuck in-flight flag.
+    // Matches the pre-queue behaviour, where write() on a closed socket was a
+    // silent no-op with no error() callback.
+    if (!m_status || !m_socket || !m_socket->is_open())
+    {
+        fail_write_queue();
+        return;
+    }
+
     boost::asio::async_write(*m_socket, boost::asio::buffer(packet->data(), packet->size()),
         [self = shared_from_this(), this, packet](const boost::system::error_code& ec, std::size_t length)
         {
             if (!ec) g_bytes_sent.fetch_add(length, std::memory_order_relaxed);
             if (ec) {
+                // A failed write kills the connection, so everything still
+                // queued behind it could never be delivered either: drop the
+                // chain and report ONCE — exactly what a single failed write
+                // did before the queue existed.
+                fail_write_queue();
                 std::shared_ptr<INetwork> strong;
                 if (!acquire_node(strong)) {
                     abort_connection();
                     return;
                 }
                 m_node->error("Socket::write error: " + ec.message(), get_addr());
+                return;
+            }
+
+            {
+                std::lock_guard<std::mutex> lock(m_write_mutex);
+                if (!m_write_queue.empty())
+                    m_write_queue.pop_front();
             }
+            // Start the NEXT composed write only from here — that ordering is
+            // the whole point of the queue. Not unbounded recursion:
+            // async_write never invokes its handler inline from the initiating
+            // call, so each do_write() returns before the next one runs.
+            do_write();
         }
     );
 }

src/core/socket.hpp

@@ -1,7 +1,9 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 #pragma once
 
+#include <deque>
 #include <map>
+#include <mutex>
 #include <vector>
 #include <string>
 #include <memory>
@@ -73,6 +75,42 @@ class Socket : public std::enable_shared_from_this<Socket>
     NetService m_addr;
     NetService m_addr_local;
 
+    // ── Outbound write queue (issue #863) ──────────────────────────────────
+    //
+    // Asio forbids initiating a second COMPOSED write on a descriptor before
+    // the first one completes: the two async_write_some continuations are
+    // serviced FIFO per descriptor, so a message needing more than one round
+    // gets bytes from the other message spliced into its middle. The peer then
+    // sees a bad length/checksum and drops the connection.
+    //
+    // Every coin's send_shares issues three back-to-back writes on one socket
+    // in a single handler turn (remember_tx -> shares -> forget_tx), and
+    // handle_version writes getaddrs/addrme before the (potentially ~32 KB)
+    // have_tx advert. Because send_shares reports its hashes as "sent" on
+    // SUBMISSION and broadcast_share never retries, a spliced frame loses that
+    // share to that peer permanently — silent PPLNS loss.
+    //
+    // Fix: buffer composed writes here and drain strictly in submission order,
+    // starting the next one only from the previous completion handler, so at
+    // most ONE composed async_write is ever in flight per socket. Same shape as
+    // StratumSession::send_response/do_write (core/stratum_server.cpp).
+    //
+    // m_write_mutex is a LEAF lock: it is only ever held around the deque/flag
+    // mutation and is never held across async_write initiation, an m_node
+    // callback, or any other lock — writes are submitted from node/compute
+    // threads as well as from the io_context thread, so the queue itself must
+    // be guarded, but no new lock-order edge is introduced.
+    //
+    // Deliberately UNCAPPED, unlike StratumSession's backlog cap: before this
+    // queue existed a stalled peer already pinned one PackStream per
+    // overlapping in-flight async_write, so the memory profile is unchanged,
+    // and dropping a queued p2p message would reintroduce exactly the silent
+    // share loss this fixes. A stuck peer is still reaped by the existing
+    // read-side error path.
+    mutable std::mutex m_write_mutex;
+    std::deque<std::shared_ptr<PackStream>> m_write_queue;
+    bool m_writing {false}; // true while a composed async_write is in flight
+
 public:
     // Global P2P traffic counters (all sockets combined)
     static inline std::atomic<uint64_t> g_bytes_recv{0};
@@ -94,6 +132,16 @@ class Socket : public std::enable_shared_from_this<Socket>
 
     void read();   // moved out-of-line; needs INetwork complete via acquire_node
 
+    // Drain step of the outbound queue: initiates the composed async_write for
+    // the front element and re-arms itself from that write's completion
+    // handler. Never called while another composed write is in flight.
+    void do_write();
+
+    // Drop every still-queued buffer and clear the in-flight flag. Used on
+    // write error / closed socket so the whole queued chain fails together,
+    // exactly as the single pre-queue write failed as a unit.
+    void fail_write_queue();
+
     void read_prefix(std::shared_ptr<Packet> packet);
 	void read_command(std::shared_ptr<Packet> packet);
 	void read_length(std::shared_ptr<Packet> packet);

src/core/test/CMakeLists.txt

@@ -15,7 +15,15 @@ if (BUILD_TESTING AND (GTest_FOUND OR GTEST_FOUND))
     # EXISTING allowlisted core_test target (never a new add_executable — that is
     # the #769 "Not Run" trap), and one KAT covers every coin because all five
     # node lanes consume this one shared helper.
-    add_executable(core_test pack_test.cpp events_test.cpp chain_test.cpp packet_test.cpp block_broadcast_test.cpp broadcast_convergence_matrix_test.cpp core_merkle_branches_test.cpp version_gate_test.cpp filesystem_test.cpp web_server_submitblock_test.cpp known_txs_eviction_test.cpp tx_advertiser_test.cpp)
+    #
+    # socket_write_queue_test.cpp: the core::Socket OUTBOUND WRITE QUEUE (#863).
+    # Drives a real loopback TCP pair with a small send buffer and submits
+    # overlapping composed writes from off-io threads; without the queue those
+    # writes splice bytes mid-message and the assertions fail. One KAT covers
+    # every coin because all five node lanes share this one socket layer. Same
+    # reasoning as above: FOLDED into the EXISTING allowlisted core_test target,
+    # never a new add_executable (the #769 "Not Run" trap).
+    add_executable(core_test pack_test.cpp events_test.cpp chain_test.cpp packet_test.cpp block_broadcast_test.cpp broadcast_convergence_matrix_test.cpp core_merkle_branches_test.cpp version_gate_test.cpp filesystem_test.cpp web_server_submitblock_test.cpp known_txs_eviction_test.cpp tx_advertiser_test.cpp socket_write_queue_test.cpp)
     target_link_libraries(core_test PRIVATE GTest::gtest_main core c2pool_merged_mining GTest::gtest)
     target_link_libraries(core_test PRIVATE c2pool_payout c2pool_hashrate ltc_coin)  # OBJECT-lib SCC direct-naming (#22/#39)