btc/dgb/bch: gate share broadcast on tx completeness + mark only what was sent

src/impl/bch/known_txs_retention.hpp

@@ -0,0 +1,181 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// Pure, unit-testable bookkeeping for the known-transaction pool that backs
+// share broadcast (remember_tx forwarding). Ported to the BCH variant from
+// src/impl/dash/known_txs_retention.hpp so the retention/eviction and
+// broadcast-gate semantics can be exercised without a live node.
+//
+// Two concerns:
+//   retain_template_txs() — the rolling-window retention with template-identity
+//     dedup (code review F1): register_template_txs is invoked per producer-job
+//     cache MISS, i.e. per (prev_share_hash, payout_script), NOT per template
+//     rotation. A ~50-payout-script cluster therefore re-registers the SAME
+//     template ~50 times on one tip; pushing each unconditionally collapses the
+//     N-slot window to a single template within seconds. Dedup by the template's
+//     tx-hash SET (its identity): a set already retained refreshes recency and
+//     consumes NO new slot, so N re-registrations of one template use ONE slot
+//     and the window holds N DISTINCT templates as intended.
+//
+//   all_txs_backable() — the canonical broadcast gate (F3): a share may be sent
+//     only when we HOLD the bytes of every referenced new-tx (so remember_tx can
+//     always forward them). This is what makes the "referenced unknown
+//     transaction" disconnect unreachable by construction, and — paired with the
+//     mint-time decline of shares whose txs are not retained — makes "every share
+//     we mint/broadcast is backable" a by-construction invariant.
+//
+//   partition_backable() — the batch form of the gate, applied by send_shares()
+//     to the shares it is about to write to one peer.
+//
+//   broadcast_and_mark() — the F2 marking policy: broadcast_share() may mark a
+//     share "already shared" only AFTER a peer actually accepted it. Marking the
+//     whole chain-walk up front (the pre-fix pattern) loses every share in a
+//     batch that send_shares then abandons — the gate skipped it, the tracker
+//     try_to_lock missed, or there were no peers — because the de-dup set then
+//     breaks the walk forever and nothing re-pushes it.
+//
+// NOTE (de-dup): the same primitives now exist per-coin (btc/dgb/bch/dash) while
+// the LTC lane hoists all_txs_backable into src/core/. Once both land, these
+// per-coin headers should be collapsed onto that core hoist.
+
+#pragma once
+
+#include <cstddef>
+#include <deque>
+#include <set>
+#include <vector>
+
+#include <core/uint256.hpp>
+
+namespace bch {
+
+// Retain the tx set of a newly-registered template in a rolling window of the
+// last `cap` DISTINCT templates, inserting its tx bytes into `known_txs`.
+//
+//   recent_sets : rolling history of distinct template tx-hash sets
+//                 (front = oldest, back = newest).
+//   known_txs   : uint256 -> tx map; needs insert_or_assign(k,v) and erase(k).
+//   hashes/txs  : parallel arrays of the template's tx hashes and bytes.
+//   cap         : number of DISTINCT templates to retain.
+//
+// A tx is erased from `known_txs` only once it has fallen out of EVERY retained
+// set. F1 dedup: if this template's tx set is already retained, its recency is
+// refreshed (moved to back) and no new slot is consumed.
+template <typename TxMap, typename Tx>
+inline void retain_template_txs(std::deque<std::set<uint256>>& recent_sets,
+                                TxMap& known_txs,
+                                const std::vector<uint256>& hashes,
+                                const std::vector<Tx>& txs,
+                                std::size_t cap)
+{
+    if (cap == 0)
+        return;
+
+    std::set<uint256> new_set(hashes.begin(), hashes.end());
+
+    // F1 dedup by template identity (the tx-hash set). If this template is
+    // already retained, refresh its recency (move to back) — no new slot, no
+    // re-eviction. This is what stops ~50 payout-script re-registrations of one
+    // template from evicting the rest of the window.
+    for (auto it = recent_sets.begin(); it != recent_sets.end(); ++it) {
+        if (*it == new_set) {
+            if (std::next(it) != recent_sets.end()) {
+                std::set<uint256> refreshed = std::move(*it);
+                recent_sets.erase(it);
+                recent_sets.push_back(std::move(refreshed));
+            }
+            return;
+        }
+    }
+
+    // Distinct template: insert its tx bytes and consume a slot.
+    const std::size_t n = std::min(hashes.size(), txs.size());
+    for (std::size_t i = 0; i < n; ++i)
+        known_txs.insert_or_assign(hashes[i], txs[i]);
+    recent_sets.push_back(std::move(new_set));
+
+    // Evict the oldest slots beyond `cap`; a tx is removed from `known_txs` only
+    // once it is absent from EVERY remaining retained set.
+    while (recent_sets.size() > cap) {
+        const std::set<uint256> evicted = std::move(recent_sets.front());
+        recent_sets.pop_front();
+        for (const auto& h : evicted) {
+            bool still_retained = false;
+            for (const auto& s : recent_sets) {
+                if (s.count(h)) { still_retained = true; break; }
+            }
+            if (!still_retained)
+                known_txs.erase(h);
+        }
+    }
+}
+
+// True iff every hash in `tx_hashes` is present in `held` (the bytes we hold,
+// e.g. m_known_txs). The canonical broadcast gate: a share is backable — safe to
+// send without risking the peer's "referenced unknown transaction" disconnect —
+// only when we can forward every referenced tx.
+template <typename Held>
+inline bool all_txs_backable(const std::vector<uint256>& tx_hashes,
+                             const Held& held)
+{
+    for (const auto& h : tx_hashes)
+        if (held.find(h) == held.end())
+            return false;
+    return true;
+}
+
+// Batch form of the F3 gate. Keeps in `shares` only the entries every one of
+// whose referenced new-tx hashes we hold, and returns how many were dropped.
+// `hashes_of(share)` yields that share's referenced new-tx hashes.
+//
+// Skipping costs no propagation: a share we downloaded is by definition already
+// on the network we fetched it from, and it will be re-offered on a later
+// broadcast once its tx bytes arrive (F2 leaves it un-marked).
+template <typename Share, typename Held, typename HashesOf>
+inline std::size_t partition_backable(std::vector<Share>& shares,
+                                      const Held& held,
+                                      HashesOf hashes_of)
+{
+    std::vector<Share> sendable;
+    sendable.reserve(shares.size());
+    std::size_t skipped = 0;
+    for (auto& share : shares) {
+        if (all_txs_backable(hashes_of(share), held))
+            sendable.push_back(std::move(share));
+        else
+            ++skipped;
+    }
+    shares.swap(sendable);
+    return skipped;
+}
+
+// F2 marking policy. Offers the batch to every peer via `send(peer)` — which
+// returns the hashes it ACTUALLY wrote to that peer — and marks in `marked`
+// only the hashes that reached at least one peer. Returns that count.
+//
+// The pre-fix code marked the whole chain-walk before calling send_shares, so a
+// batch abandoned by every peer (gate skip, try_to_lock miss, empty peer set)
+// stayed marked forever: the next broadcast_share walk breaks on the first
+// marked hash and the share is never re-pushed. That is silent, permanent share
+// loss with no retry path — a PPLNS-credit loss for whoever mined it.
+template <typename MarkedSet, typename Peers, typename SendFn>
+inline std::size_t broadcast_and_mark(MarkedSet& marked, Peers& peers,
+                                      const std::vector<uint256>& to_send,
+                                      SendFn send)
+{
+    if (to_send.empty())
+        return 0;
+
+    // NOTE: `to_send` is deliberately NOT what gets marked. Marking the offered
+    // batch here is precisely the pre-fix bug this signature exists to make
+    // un-writable by accident — only `send`'s report may be marked.
+    std::set<uint256> actually_sent;
+    for (auto& entry : peers) {
+        const std::vector<uint256> sent = send(entry.second);
+        actually_sent.insert(sent.begin(), sent.end());
+    }
+    for (const auto& h : actually_sent)
+        marked.insert(h);
+    return actually_sent.size();
+}
+
+} // namespace bch

src/impl/bch/node.cpp

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 #include "node.hpp"
+#include "known_txs_retention.hpp"  // bch::all_txs_backable / partition_backable / broadcast_and_mark (F2/F3)
 
 #include "share_tx_refs.hpp"    // bch::new_tx_hashes -- uniform send-side probe (#905)
 
@@ -684,8 +685,13 @@ std::vector<bch::ShareType> NodeImpl::handle_get_share(std::vector<uint256> hash
 	return shares;
 }
 
-void NodeImpl::send_shares(peer_ptr peer, const std::vector<uint256>& share_hashes)
+std::vector<uint256> NodeImpl::send_shares(peer_ptr peer,
+                                           const std::vector<uint256>& share_hashes)
 {
+    // Returns the hashes of the shares actually WRITTEN to this peer (F2). The
+    // caller marks only those as broadcast, so a share skipped here — by the
+    // try_to_lock miss below or by the F3 tx-completeness gate — is re-offered
+    // on the next broadcast instead of being silently dropped forever.
     // try_to_lock per the architectural rule (node.hpp:67) — see freeze
     // analysis in handle_get_share above.  If we can't acquire NOW, skip
     // this batch.  The shares are still in our chain; the next broadcast
@@ -697,7 +703,7 @@ void NodeImpl::send_shares(peer_ptr peer, const std::vector<uint256>& share_hash
         if (defer_log++ % 50 == 0)
             LOG_INFO << "[send_shares] tracker busy — skipping send to "
                      << peer->addr().to_string() << " (will retry next cycle)";
-        return;
+        return {};
     }
 
     // Collect shares that exist in our chain (skip rejected)
@@ -717,7 +723,51 @@ void NodeImpl::send_shares(peer_ptr peer, const std::vector<uint256>& share_hash
     }
 
     if (shares.empty())
-        return;
+        return {};
+
+    // --- F3 tx-completeness broadcast gate --------------------------------
+    // A share whose referenced new-tx BYTES we do not hold cannot be backed by
+    // the remember_tx forward below, and canonical p2pool drops the connection
+    // on "referenced unknown transaction" (p2p.py) — which isolates us from the
+    // sharechain and orphans our shares. Shares pulled via sharereply arrive
+    // WITHOUT their tx bytes, so this is reachable in normal operation.
+    //
+    // Gate on the BYTES in m_known_txs, not on the peer's have_tx advert: that
+    // advert can be stale (the peer may have dropped the tx) and sending
+    // hash-only for a tx the peer no longer holds is exactly the disconnect.
+    // m_remote_txs is used ONLY below to choose the hash-vs-bytes encoding.
+    //
+    // CONSENSUS-NEUTRAL: this decides only WHETHER we broadcast. Share bytes,
+    // minting and payout are untouched; a skipped share stays in our chain and
+    // is re-offered once its txs arrive (send_shares reports it as unsent, so
+    // broadcast_share leaves it un-marked).
+    {
+        const size_t skipped_incomplete = partition_backable(
+            shares, m_known_txs, [](ShareType& share) {
+                std::vector<uint256> hashes;
+                share.invoke([&](auto* obj) {
+                    // #880: route through the bch::new_tx_hashes SSOT (#913)
+                    // instead of the dead flat probe -- v17/v33 nest the list in
+                    // m_tx_info, v34+ carry none. The old probe was FALSE for
+                    // every variant, so this F3 gate never withheld a share.
+                    if (const auto* new_txs = bch::new_tx_hashes(obj))
+                        hashes.assign(new_txs->begin(), new_txs->end());
+                });
+                return hashes;
+            });
+        if (skipped_incomplete > 0)
+        {
+            static int skip_log = 0;
+            if (skip_log++ % 20 == 0)
+                LOG_INFO << "[send_shares] skipped " << skipped_incomplete
+                         << " share(s) whose new-tx bytes we do not hold "
+                            "(downloaded via sharereply / evicted) to "
+                         << peer->addr().to_string()
+                         << " — avoids canonical 'unknown transaction' disconnect";
+        }
+        if (shares.empty())
+            return {};
+    }
 
     // Collect transactions that the peer doesn't know about
     std::set<uint256> needed_txs;
@@ -749,6 +799,8 @@ void NodeImpl::send_shares(peer_ptr peer, const std::vector<uint256>& share_hash
         std::vector<uint256> known_hashes;   // hashes in peer's remote set
         std::vector<coin::MutableTransaction> full_txs;  // full txs otherwise
 
+        size_t missing = 0;
+
         for (const auto& th : needed_txs)
         {
             if (peer->m_remote_txs.count(th))
@@ -760,9 +812,20 @@ void NodeImpl::send_shares(peer_ptr peer, const std::vector<uint256>& share_hash
                 auto it = m_known_txs.find(th);
                 if (it != m_known_txs.end())
                     full_txs.emplace_back(it->second);
+                else
+                    ++missing;  // unreachable after the gate above — see below
             }
         }
 
+        // Defence in depth: the gate should already have removed every share
+        // with an unheld tx, so this must stay at zero. If it ever fires, the
+        // gate's view of the referenced hashes disagrees with needed_txs and
+        // the peer may drop these shares — make that loud rather than silent.
+        if (missing > 0)
+            LOG_WARNING << "[send_shares] " << missing << " referenced tx(s) not in "
+                           "m_known_txs after the completeness gate — peer may drop "
+                           "these shares (share/tx hash accounting mismatch?)";
+
         if (!known_hashes.empty() || !full_txs.empty())
         {
             auto rtx_msg = message_remember_tx::make_raw(known_hashes, full_txs);
@@ -799,6 +862,13 @@ void NodeImpl::send_shares(peer_ptr peer, const std::vector<uint256>& share_hash
 
     LOG_INFO << "[Pool] Sent " << shares.size() << " shares (+" << needed_txs.size()
              << " txs) to " << peer->addr().to_string();
+
+    // F2: report exactly which shares reached this peer.
+    std::vector<uint256> sent;
+    sent.reserve(shares.size());
+    for (auto& share : shares)
+        sent.push_back(share.hash());
+    return sent;
 }
 
 void NodeImpl::broadcast_share(const uint256& share_hash)
@@ -821,7 +891,14 @@ void NodeImpl::broadcast_share(const uint256& share_hash)
         return;
     }
 
-    // Walk the chain back from share_hash, collecting un-broadcast shares
+    // Walk the chain back from share_hash, collecting un-broadcast shares.
+    //
+    // F2: do NOT mark them shared here. send_shares can abandon the whole batch
+    // — the tx-completeness gate skipped it, its tracker try_to_lock missed, or
+    // there are zero peers — and a share marked before that never gets another
+    // chance: the next walk breaks on the first marked hash, so the tip is
+    // permanently withheld from the sharechain and its PPLNS credit is lost.
+    // We mark only what actually reached >= 1 peer, below.
     std::vector<uint256> to_send;
     int32_t height = m_chain->get_height(share_hash);
     int32_t walk = std::min(height, 5);
@@ -832,18 +909,23 @@ void NodeImpl::broadcast_share(const uint256& share_hash)
             break;
         if (m_rejected_share_hashes.count(hash))
             continue; // skip shares previously rejected by peers
-        m_shared_share_hashes.insert(hash);
         to_send.push_back(hash);
     }
 
     if (to_send.empty())
         return;
 
     auto now = std::chrono::steady_clock::now();
-    for (auto& [nonce, peer] : m_peers) {
-        send_shares(peer, to_send);
-        m_last_broadcast_to[peer->addr()] = {to_send, now};
-    }
+    broadcast_and_mark(m_shared_share_hashes, m_peers, to_send, [&](peer_ptr& peer) {
+        std::vector<uint256> sent = send_shares(peer, to_send);
+        // Record what we ACTUALLY wrote, not what we offered: a peer that drops
+        // within 10s marks this record's hashes rejected, and a rejected hash is
+        // never re-broadcast. Blaming a share we withheld would turn a transient
+        // gate skip into permanent loss.
+        if (!sent.empty())
+            m_last_broadcast_to[peer->addr()] = {sent, now};
+        return sent;
+    });
 }
 
 void NodeImpl::notify_local_share(const uint256& share_hash)
@@ -983,6 +1065,20 @@ uint256 NodeImpl::advertised_best_share()
     return uint256::ZERO;
 }
 
+// ROOT-2 re-advert. Unlike ltc/dgb (which have a dedicated readvertise that
+// bypasses the de-dup set), this delegates to broadcast_share, so the walk still
+// breaks on the first hash already in m_shared_share_hashes.
+//
+// Interaction with the F2 marking change: marking is now a strict SUBSET of what
+// the pre-fix code marked (only hashes a peer actually accepted). The walk breaks
+// later or in the same place, never earlier, so this re-advert re-pushes a
+// superset of what it used to — monotonically better, no regression. It is
+// strictly better in the case that motivated ROOT-2: head shares minted while no
+// peer was connected are no longer marked, so a peer that handshook during the
+// empty window now receives them. It does NOT fully close ROOT-2 here — head
+// shares already accepted by some OTHER peer stay marked and the walk still
+// breaks, exactly as before. Closing that needs the ltc/dgb-style de-dup-bypass
+// readvertise; pre-existing gap, out of scope for this change.
 void NodeImpl::readvertise_best()
 {
     if (m_peers.empty())

src/impl/bch/node.hpp

@@ -605,9 +605,13 @@ class NodeImpl : public pool::SharechainNode<bch::Config, bch::ShareChain, bch::
     /// Set the software version string announced to peers in P2P version messages.
     void set_software_version(std::string ver) { m_software_version = std::move(ver); }
 
-    /// Send a set of shares (with any needed txs) to a single peer.
-    /// Skips shares that originated from that peer.
-    void send_shares(peer_ptr peer, const std::vector<uint256>& share_hashes);
+    /// Send a set of shares (with any needed txs) to a single peer, and return
+    /// the subset actually written (F2). Shares dropped by the tx-completeness
+    /// gate, or the whole batch when the tracker lock is unavailable, are
+    /// reported as NOT sent, so the caller leaves them un-marked and retries
+    /// them on the next broadcast.
+    std::vector<uint256> send_shares(peer_ptr peer,
+                                     const std::vector<uint256>& share_hashes);
 
     /// Broadcast a locally-generated (or newly-received) share to all peers.
     void broadcast_share(const uint256& share_hash);