bch(M5): reorg re-request connector — daemon-assembly integration

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;
@@ -198,6 +200,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
@@ -342,6 +359,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 +429,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

@@ -83,6 +83,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 +103,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.)

src/impl/bch/test/reorg_rerequest_test.cpp

@@ -0,0 +1,169 @@
+// ---------------------------------------------------------------------------
+// bch::coin reorg re-request test (M5 full-block body).
+//
+// The BlockConnector reconnects a reorg's new branch from its bounded
+// remembered-block ring. When the chain advances to a new tip whose
+// intermediate new-branch blocks have NOT been delivered as full blocks yet
+// (the normal headers-first case: headers race ahead of block downloads) or
+// when a reorg runs deeper than the ring retains, recall_block() returns null
+// and the old behaviour was to log a warning and strand the UTXO at the fork
+// with NO recovery. This test pins the now-live recovery leg:
+//
+//   * a set_block_requester() sink, when wired, is handed exactly the
+//     not-yet-connected new-branch hashes (fork->tip order) so they can be
+//     re-getdata'd (BlockDownloadWindow::enqueue) and the reorg completes on
+//     re-delivery -- instead of silently holding at the fork;
+//   * the fork-point block stays connected (it is at/below the fork, never
+//     disconnected) so the UTXO view is left consistent, not corrupted;
+//   * with NO sink wired the connector still falls back to warn-and-hold
+//     (no crash, no throw) -- the cold-start / unwired contract holds.
+//
+// Build posture mirrors utxo_connect_test: header-only over coin/*.hpp +
+// <core/*> plus coin/transaction.cpp; NO impl_bch coin lib (per-coin isolation
+// stays clean). All heights sit far below the mainnet ASERT anchor so the
+// header chain trusts difficulty structurally, and peer_tip is set high so PoW
+// is skipped -- this builds a deterministic 3-block fork with no PoW grinding.
+// p2pool-merged-v36 surface: NONE (local block-connect / download plumbing).
+// ---------------------------------------------------------------------------
+
+#include <cstdint>
+#include <iostream>
+#include <vector>
+
+#include "../coin/block.hpp"
+#include "../coin/block_connector.hpp"
+#include "../coin/header_chain.hpp"
+#include "../coin/mempool.hpp"
+#include "../coin/transaction.hpp"
+
+#include <core/coin/utxo_view_cache.hpp>
+
+namespace {
+
+int failures = 0;
+#define CHECK(cond) do { if (!(cond)) { \
+    std::cerr << "FAIL: " #cond " @ line " << __LINE__ << "\n"; ++failures; } } while (0)
+
+bch::coin::MutableTransaction make_cb(const char* prev_hex, int64_t out_value) {
+    bch::coin::MutableTransaction tx;
+    bch::coin::TxIn in;
+    in.prevout.hash.SetHex(prev_hex);
+    in.prevout.index = 0;
+    in.sequence = 0xffffffff;
+    tx.vin.push_back(in);
+    bch::coin::TxOut out;
+    out.value = out_value;
+    tx.vout.push_back(out);
+    return tx;
+}
+
+// A header building on `prev` with a distinct nonce (=> distinct hash).
+bch::coin::BlockType make_block(const uint256& prev, uint32_t nonce) {
+    bch::coin::BlockType b;
+    b.m_version = 0x20000000;
+    b.m_previous_block = prev;
+    b.m_merkle_root.SetHex(
+        "1111111111111111111111111111111111111111111111111111111111111111");
+    b.m_timestamp = 1700000000 + nonce;
+    b.m_bits = 0x1d00ffff;
+    b.m_nonce = nonce;
+    return b;
+}
+
+const char* OUTPOINT_A = "aa00000000000000000000000000000000000000000000000000000000000000";
+
+// Header chain whose fast-start checkpoint IS block1's hash at height H, so the
+// connector sees block1 as the tip without fighting PoW.
+bch::coin::HeaderChain make_chain(const uint256& tip_hash, uint32_t H) {
+    using namespace bch::coin;
+    BCHChainParams params = BCHChainParams::mainnet();
+    params.fast_start_checkpoint = BCHChainParams::Checkpoint{H, tip_hash};
+    return HeaderChain(params);
+}
+
+} // namespace
+
+int main() {
+    using namespace bch::coin;
+
+    // Heights far below the mainnet ASERT anchor (661647) => difficulty trusted
+    // structurally; peer_tip high => PoW skipped. Deterministic 3-block fork.
+    const uint32_t H = 1000;
+
+    uint256 root_prev; root_prev.SetHex(
+        "00000000000000000001a2b3c4d5e6f700000000000000000000000000000000");
+    BlockType block1 = make_block(root_prev, /*nonce=*/1);
+    block1.m_txs.push_back(make_cb(OUTPOINT_A, 100));   // coinbase-position tx
+    const uint256 h1 = block_hash(static_cast<const BlockHeaderType&>(block1));
+
+    BlockType block2 = make_block(h1, /*nonce=*/2);     // intermediate, NEVER delivered
+    const uint256 h2 = block_hash(static_cast<const BlockHeaderType&>(block2));
+    BlockType block3 = make_block(h2, /*nonce=*/3);      // new tip, delivered
+    const uint256 h3 = block_hash(static_cast<const BlockHeaderType&>(block3));
+    CHECK(h1 != h2); CHECK(h2 != h3); CHECK(h1 != h3);
+
+    // ---- 1) Sink wired: missing new-branch hashes are re-requested ---------
+    {
+        HeaderChain chain = make_chain(h1, H);
+        CHECK(chain.init());
+        chain.set_peer_tip_height(H + 100000);   // skip PoW for these heights
+
+        core::coin::UTXOViewCache utxo(nullptr);
+        Mempool pool;
+
+        BlockConnector conn(chain, pool);
+        conn.set_utxo(&utxo);
+
+        std::vector<uint256> requested;
+        int sink_calls = 0;
+        conn.set_block_requester([&](const std::vector<uint256>& v) {
+            requested = v; ++sink_calls;
+        });
+
+        // Cold-start forward-extend: connect block1 (the checkpoint tip).
+        conn.on_full_block(block1);
+
+        // Chain advances to block3 via headers only; block2's full block never
+        // arrives, so it is absent from the connector's remembered ring.
+        CHECK(chain.add_header(static_cast<const BlockHeaderType&>(block2)));
+        CHECK(chain.add_header(static_cast<const BlockHeaderType&>(block3)));
+        auto tip = chain.tip();
+        CHECK(tip.has_value() && tip->block_hash == h3);   // block3 is best tip
+
+        // Deliver block3: reorg fork=block1 -> [block2, block3]; block2 missing.
+        conn.on_full_block(block3);
+
+        CHECK(sink_calls == 1);
+        CHECK(requested.size() == 2);
+        if (requested.size() == 2) {
+            CHECK(requested[0] == h2);   // fork->tip order: missing intermediate first
+            CHECK(requested[1] == h3);
+        }
+    }
+
+    // ---- 2) No sink wired: warn-and-hold, no crash/throw -------------------
+    {
+        HeaderChain chain = make_chain(h1, H);
+        CHECK(chain.init());
+        chain.set_peer_tip_height(H + 100000);
+
+        core::coin::UTXOViewCache utxo(nullptr);
+        Mempool pool;
+
+        BlockConnector conn(chain, pool);
+        conn.set_utxo(&utxo);
+        // NO set_block_requester()
+
+        conn.on_full_block(block1);
+        CHECK(chain.add_header(static_cast<const BlockHeaderType&>(block2)));
+        CHECK(chain.add_header(static_cast<const BlockHeaderType&>(block3)));
+        conn.on_full_block(block3);      // must not throw / crash
+    }
+
+    if (failures == 0) {
+        std::cout << "reorg_rerequest_test: ALL PASS\n";
+        return 0;
+    }
+    std::cerr << "reorg_rerequest_test: " << failures << " FAILURE(S)\n";
+    return 1;
+}