bch(M5): wire embedded P2P leg in production run() — close won-block P2P degradation

.github/workflows/build.yml

@@ -284,7 +284,7 @@ jobs:
                     bch_coinbase_kat_segwit_predicate_test \
                     bch_coinbase_kat_bytevector_test \
                     bch_block_connector_test bch_block_ordering_test \
-                    bch_utxo_connect_test bch_reorg_connect_test \
+                    bch_utxo_connect_test bch_reorg_connect_test bch_reorg_rerequest_test \
                     bch_compact_block_connector_test \
                     bch_embedded_daemon_assembly_test \
                     bch_embedded_getwork_test bch_embedded_seam_workview_test \
@@ -358,7 +358,7 @@ jobs:
                     bch_coinbase_kat_segwit_predicate_test \
                     bch_coinbase_kat_bytevector_test \
                     bch_block_connector_test bch_block_ordering_test \
-                    bch_utxo_connect_test bch_reorg_connect_test \
+                    bch_utxo_connect_test bch_reorg_connect_test bch_reorg_rerequest_test \
                     bch_compact_block_connector_test \
                     bch_embedded_daemon_assembly_test \
                     bch_embedded_getwork_test bch_embedded_seam_workview_test \

src/impl/bch/coin/block_connector.hpp

@@ -63,6 +63,7 @@
 #include <deque>
 #include <memory>
 #include <optional>
+#include <functional>
 #include <unordered_map>
 #include <vector>
 
@@ -89,6 +90,18 @@ class BlockConnector {
     /// is safe.
     void set_utxo(core::coin::UTXOViewCache* u) { m_utxo = u; }
 
+    /// Optional re-request sink. A reorg deeper than the remembered-block
+    /// ring (or a headers-first reorg whose intermediate new-branch blocks
+    /// have not been downloaded yet) can no longer reconnect from cache; the
+    /// missing new-branch hashes are handed here -- wire it to
+    /// BlockDownloadWindow::enqueue() so they are re-getdata'd and the reorg
+    /// completes on their re-delivery, instead of stranding the UTXO view at
+    /// the fork. Left unset the connector falls back to the warn-and-hold
+    /// behaviour (UTXO safely parked at the fork until an external resync).
+    void set_block_requester(std::function<void(const std::vector<uint256>&)> req) {
+        m_request_blocks = std::move(req);
+    }
+
     /// Wire to a node's full_block event. Retained internally, torn down on
     /// destruction (or detach()). Call once, after node + chain + pool exist.
     void attach(bch::interfaces::Node& node) {
@@ -147,6 +160,12 @@ class BlockConnector {
 
     bool is_attached() const { return static_cast<bool>(m_sub); }
 
+    /// True once a re-request sink has been wired (set_block_requester). When
+    /// false a deep reorg falls back to warn-and-hold at the fork; when true the
+    /// missing new-branch bodies are re-getdata'd. Exposed so the daemon-assembly
+    /// path can verify the wiring without a live reorg.
+    bool has_block_requester() const { return static_cast<bool>(m_request_blocks); }
+
     /// Ingest a BIP152 compact block. Reconstructs the full block from the
     /// prefilled txs + the mempool (BCH: short IDs keyed on txid == wtxid). On a
     /// complete reconstruction the block is driven straight through the normal
@@ -314,9 +333,22 @@ class BlockConnector {
         for (auto it = connect_path.rbegin(); it != connect_path.rend(); ++it) {
             const BlockType* blk = recall_block(*it);
             if (!blk) {
+                // Reorg deeper than the remembered ring (or a headers-first
+                // reorg whose intermediate new-branch blocks were not yet
+                // downloaded): this block and every block still above it on
+                // the path are absent from the cache. Re-request them for
+                // getdata so the reorg completes on their re-delivery instead
+                // of stranding the UTXO at the fork. The sink (enqueue())
+                // dedupes, so an overlapping request never double-getdata's.
+                // `it .. rend()` is exactly the not-yet-connected tail, in
+                // fork->tip connect order.
+                std::vector<uint256> missing(it, connect_path.rend());
                 LOG_WARNING << "[EMB-BCH] reorg: new-branch block "
                                   << it->GetHex().substr(0, 16)
-                                  << "... not retained -- UTXO left at fork, awaiting resync";
+                                  << "... not retained -- re-requesting " << missing.size()
+                                  << " block(s), UTXO held at fork until re-delivery";
+                if (m_request_blocks)
+                    m_request_blocks(missing);
                 return;
             }
             auto e = m_chain.get_header(*it);
@@ -327,6 +359,9 @@ class BlockConnector {
     HeaderChain&                 m_chain;
     Mempool&                     m_pool;
     core::coin::UTXOViewCache*   m_utxo {nullptr};   // optional UTXO view
+    // Re-request sink for reorg blocks absent from the remembered ring
+    // (see set_block_requester). Unset -> warn-and-hold at the fork.
+    std::function<void(const std::vector<uint256>&)> m_request_blocks;
 
     uint256                      m_connected_tip {}; // null = nothing applied yet
     std::vector<ConnectedEntry>  m_undo_stack;       // applied best-chain blocks

src/impl/bch/coin/embedded_daemon.hpp

@@ -56,6 +56,7 @@
 
 #include "header_chain.hpp"     // HeaderChain
 #include "mempool.hpp"          // Mempool
+#include "block_connector.hpp" // BlockConnector (full-block UTXO/mempool reorg path)
 #include "template_builder.hpp" // EmbeddedCoinNode
 #include "node.hpp"             // coin::Node<Config> (interfaces::Node source)
 #include "coin_node.hpp"        // CoinNode (core::coin::ICoinNode seam)
@@ -86,7 +87,8 @@ class EmbeddedDaemon {
           m_pool(),
           m_embedded(m_chain, m_pool, config->m_testnet),
           m_node(context, config),
-          m_abla(config->m_testnet, anchor_height, m_chain) {}
+          m_abla(config->m_testnet, anchor_height, m_chain),
+          m_connector(m_chain, m_pool) {}
 
     EmbeddedDaemon(const EmbeddedDaemon&) = delete;
     EmbeddedDaemon& operator=(const EmbeddedDaemon&) = delete;
@@ -101,9 +103,11 @@ class EmbeddedDaemon {
         assemble();                   // network-free seam + ABLA wiring (see below)
         wire_chain_ingest();          // new_headers --> HeaderChain (advances synced height)
         pin_cold_start_anchor();      // operator-APPROVED VM300 anchor (decisions@ 2026-06-18); floor-equivalent
+        const bool p2p_up = maybe_start_p2p(); // arm won-block P2P relay leg + connector re-request sink (gated on configured peer)
         LOG_INFO << "[EMB-BCH] embedded daemon up: embedded-primary work source,"
                  << " external BCHN-RPC fallback retained, ABLA loop closed"
-                 << " (cold-start anchor pinned @" << BchnAnchorRecord::height << ").";
+                 << " (cold-start anchor pinned @" << BchnAnchorRecord::height << "),"
+                 << " P2P relay leg " << (p2p_up ? "LIVE" : "OFF (no peer configured -> RPC-only)") << ".";
     }
 
     /// READ-ONLY IBD evidence harness entry (drives the --ibd run-loop in
@@ -198,6 +202,21 @@ class EmbeddedDaemon {
             return;                   // already assembled; idempotent no-op
         m_abla.wire(m_node, m_embedded);
         m_coin_node = std::make_unique<CoinNode>(&m_embedded, m_node.rpc());
+
+        // Drive the full-block UTXO/mempool reorg-reconciliation path and wire its
+        // deep-reorg re-request sink to the P2P block-download window: a reorg
+        // deeper than the connector's remembered-block ring re-getdata's the
+        // missing new-branch bodies through the bounded/deduping IBD window
+        // instead of stranding the UTXO view at the fork. m_node.p2p() is null
+        // until start_p2p(), so the sink no-ops safely when assemble() runs
+        // network-free. attach() subscribes the connector to full_block (the
+        // m_coin_node guard above makes this run exactly once).
+        m_connector.set_block_requester(
+            [this](const std::vector<uint256>& hashes) {
+                if (auto* p2p = m_node.p2p())
+                    p2p->request_block_downloads(hashes);
+            });
+        m_connector.attach(m_node);
     }
 
     /// Wire NodeP2P's IBD getheaders continuation to the authoritative
@@ -213,6 +232,32 @@ class EmbeddedDaemon {
             p2p->set_locator_provider([this]{ return m_chain.get_locator(); });
     }
 
+    /// Bring up the embedded BCH P2P transport against the configured peer
+    /// (coin()->m_p2p.address) when one is set, then bind the HeaderChain
+    /// locator provider so any IBD continuation anchors at the learned tip.
+    /// Returns true if the transport was started, false when no peer is
+    /// configured (port == 0) -- in which case the daemon stays strictly
+    /// RPC-only (offline/no-peer contract preserved).
+    ///
+    /// Closes the won-block P2P-leg gap: broadcast_won_block submit_block_p2p*
+    /// calls no-op while m_node.p2p() is null, so EVERY won block degraded to
+    /// the RPC-only submitblock leg; likewise the BlockConnector
+    /// request_block_downloads re-request sink (wired in assemble) stayed
+    /// dormant with no download window to issue getdata on. Both go live the
+    /// instant the transport exists. The external BCHN-RPC fallback (init_rpc,
+    /// already brought up by run) is untouched -- this only ADDS the P2P leg.
+    /// A P2P connect issues no qm/control op, so VM300 bchn-bch stays read-only.
+    /// p2pool-merged-v36 surface: NONE (transport wiring only -- no PoW/share/
+    /// coinbase/PPLNS/WorkData-shape change).
+    bool maybe_start_p2p() {
+        const auto& peer = m_config->coin()->m_p2p.address;
+        if (peer.port() == 0)             // no peer configured -> RPC-only
+            return false;
+        m_node.start_p2p(peer);           // embedded BCHN P2P relay/IBD transport
+        bind_locator_provider();          // HeaderChain back-off locator for ContinueSync
+        return true;
+    }
+
     /// Drive the authoritative HeaderChain from the live P2P header stream.
     /// The P2P front-end (NodeP2P) parses `headers` messages and fires
     /// new_headers; until this subscription existed NOTHING advanced m_chain
@@ -342,6 +387,7 @@ class EmbeddedDaemon {
     CoinNode&           coin_node()      { return *m_coin_node; }
     bool                seam_ready() const { return m_coin_node != nullptr; }
     AblaRuntime&        abla()           { return m_abla; }
+    BlockConnector&     connector()      { return m_connector; }
     HeaderChain&        chain()          { return m_chain; }
     Mempool&            mempool()        { return m_pool; }
     bool                is_wired() const { return m_abla.is_wired(); }
@@ -411,6 +457,10 @@ class EmbeddedDaemon {
     EmbeddedCoinNode m_embedded; // in-process work source
     Node<config_t>   m_node;     // P2P + external-RPC fallback; full_block source
     AblaRuntime      m_abla;     // owns tracker + feed; wired in run()
+    // full_block -> header connect -> best-chain-gated UTXO/mempool reconcile;
+    // attached in assemble(). Declared after m_node so its full_block
+    // subscription tears down (dtor detach) before the event source m_node dies.
+    BlockConnector   m_connector;
     // new_headers -> m_chain.add_headers() subscription (headers-first IBD).
     std::shared_ptr<EventDisposable> m_headers_sub;
     // Built in run() once m_node.rpc() is live; binds raw ptrs to m_embedded

src/impl/bch/coin/p2p_node.hpp

@@ -366,6 +366,20 @@ class NodeP2P : public core::ICommunicator, public core::INetwork, public core::
         }
     }
 
+    /// Re-request a batch of block bodies through the bounded, deduping,
+    /// reissue-accounted download window -- the wiring target for
+    /// BlockConnector::set_block_requester (deep-reorg re-request recovery).
+    /// Routed through m_block_dl (NOT request_full_block) so re-getdata'd bodies
+    /// stay under the same in-flight bound + timeout/eviction accounting as IBD;
+    /// the window dedupes against anything already requested. No-op offline
+    /// (drain_block_window is a no-op without a peer) -- safe to call before P2P.
+    void request_block_downloads(const std::vector<uint256>& hashes)
+    {
+        if (hashes.empty()) return;
+        m_block_dl.enqueue(hashes);
+        drain_block_window();
+    }
+
     /// Request a block via plain MSG_BLOCK (0x02) getdata.
     void request_block(const uint256& block_hash)
     {

src/impl/bch/test/CMakeLists.txt

@@ -103,6 +103,24 @@ if(BUILD_TESTING)
         btclibs)
     add_test(NAME bch_reorg_connect_test COMMAND bch_reorg_connect_test)
 
+    # reorg_rerequest_test: M5 reorg recovery leg -- the COMPLEMENT of
+    # reorg_connect_test. When the new branch's intermediate blocks are NOT
+    # in the remembered ring (headers-first race, or a reorg deeper than the
+    # ring retains), the connector hands the missing new-branch hashes to a
+    # set_block_requester() sink (wired to BlockDownloadWindow::enqueue) so
+    # they are re-getdata'd and the reorg completes on re-delivery, instead
+    # of stranding the UTXO at the fork. Same link set + transaction.cpp TU;
+    # no impl_bch coin lib -> per-coin isolation holds.
+    add_executable(bch_reorg_rerequest_test
+        reorg_rerequest_test.cpp
+        ${CMAKE_SOURCE_DIR}/src/impl/bch/coin/transaction.cpp)
+    target_include_directories(bch_reorg_rerequest_test PRIVATE ${CMAKE_SOURCE_DIR}/src)
+    target_link_libraries(bch_reorg_rerequest_test PRIVATE
+        core pool sharechain
+        c2pool_payout c2pool_merged_mining c2pool_hashrate c2pool_storage c2pool_difficulty
+        btclibs)
+    add_test(NAME bch_reorg_rerequest_test COMMAND bch_reorg_rerequest_test)
+
     # compact_block_connector_test: M5 (b) -- BIP152 compact-block depth on the
     # connector seam. on_compact_block reconstructs from prefilled + mempool; a
     # missing tx drives a getblocktxn round that on_block_txn closes out before

src/impl/bch/test/embedded_daemon_assembly_test.cpp

@@ -38,6 +38,7 @@
 
 #include <boost/asio/io_context.hpp>
 
+#include <core/netaddress.hpp>   // NetService (no-peer RPC-only contract)
 #include "../coin/embedded_daemon.hpp"
 
 namespace {
@@ -55,7 +56,7 @@ struct TestConfig {
     // Duck-typed coin()->m_p2p.prefix: NodeP2P<Config>::get_prefix() is a virtual
     // override (eagerly instantiated via the vtable) and reads it. Empty here --
     // assemble() never sends a P2P message, so the value is inert.
-    struct P2P { std::vector<std::byte> prefix; };
+    struct P2P { std::vector<std::byte> prefix; NetService address; };
     struct Coin { P2P m_p2p; };
     Coin m_coin;
     const Coin* coin() const { return &m_coin; }
@@ -83,6 +84,17 @@ int main() {
     CHECK(daemon.coin_node().is_embedded());     // embedded work source = primary
     CHECK(!daemon.coin_node().has_rpc());         // RPC fallback absent offline
 
+    // 2b) FULL-BLOCK REORG PATH WIRED. assemble() now also instantiates +
+    //     attaches the daemon-owned BlockConnector to full_block (so every
+    //     received block drives header connect + best-chain-gated UTXO/mempool
+    //     reconciliation) and binds its deep-reorg re-request sink to the P2P
+    //     download window. has_block_requester()=true proves a reorg deeper than
+    //     the remembered-block ring re-getdata's the missing new-branch bodies
+    //     instead of stranding the UTXO view at the fork; the sink itself no-ops
+    //     offline (m_node.p2p() null until start_p2p()), so this is network-free.
+    CHECK(daemon.connector().is_attached());
+    CHECK(daemon.connector().has_block_requester());
+
     // 3) The seam is backed by the daemon's REAL EmbeddedCoinNode, not a fake:
     //    a fresh, un-init'd HeaderChain reports NOT synced (FakeEmbedded=true).
     CHECK(!daemon.embedded().is_synced());
@@ -92,6 +104,8 @@ int main() {
     daemon.assemble();
     CHECK(&daemon.coin_node() == before);
     CHECK(daemon.seam_ready());
+    CHECK(daemon.connector().is_attached());      // connector wiring stable across re-assemble
+    CHECK(daemon.connector().has_block_requester());
 
     // 5) Cold-start anchor DRY RUN: record-only, VM300 untouched, floor no-op.
     //    (Just exercises the path -- it logs and must not throw or mutate.)
@@ -109,6 +123,17 @@ int main() {
     CHECK(daemon.abla().tracker().budget_for_tip(Rec::height)
           == Rec::abla_blocksizelimit);          // floor-equivalent: 32 MB
 
+    // 7) NO-PEER RPC-ONLY CONTRACT (maybe_start_p2p offline contract). With no
+    //    P2P peer configured (default NetService, port==0) maybe_start_p2p() is a
+    //    no-op: returns false and leaves node().has_p2p() false, so the daemon
+    //    stays strictly on the external BCHN-RPC submitblock leg -- the won-block
+    //    P2P relay leg and the connector re-request sink stay correctly dormant.
+    //    This is exactly the path run() takes when coin()->m_p2p.address is unset,
+    //    and it must never spin up a transport (no qm/control op -> VM300 read-only).
+    CHECK(config.m_coin.m_p2p.address.port() == 0); // precondition: no peer set
+    CHECK(!daemon.maybe_start_p2p());                // no peer -> no-op, returns false
+    CHECK(!daemon.node().has_p2p());                 // P2P leg stays dormant (RPC-only)
+
     if (failures == 0) {
         std::cout << "embedded_daemon_assembly_test: ALL PASS\n";
         return 0;