[WIP] new rebroadcast logic

src/miner.cpp

@@ -45,13 +45,15 @@ int64_t UpdateTime(CBlockHeader* pblock, const Consensus::Params& consensusParam
 BlockAssembler::Options::Options() {
     blockMinFeeRate = CFeeRate(DEFAULT_BLOCK_MIN_TX_FEE);
     nBlockMaxWeight = DEFAULT_BLOCK_MAX_WEIGHT;
+    nMinTxAge = 0;
 }
 
 BlockAssembler::BlockAssembler(const CChainParams& params, const Options& options) : chainparams(params)
 {
     blockMinFeeRate = options.blockMinFeeRate;
     // Limit weight to between 4K and MAX_BLOCK_WEIGHT-4K for sanity:
     nBlockMaxWeight = std::max<size_t>(4000, std::min<size_t>(MAX_BLOCK_WEIGHT - 4000, options.nBlockMaxWeight));
+    nMinTxAge = options.nMinTxAge;
 }
 
 static BlockAssembler::Options DefaultOptions()
@@ -66,6 +68,8 @@ static BlockAssembler::Options DefaultOptions()
     } else {
         options.blockMinFeeRate = CFeeRate(DEFAULT_BLOCK_MIN_TX_FEE);
     }
+    options.nMinTxAge = 0;
+
     return options;
 }
 
@@ -271,6 +275,7 @@ int BlockAssembler::UpdatePackagesForAdded(const CTxMemPool::setEntries& already
 bool BlockAssembler::SkipMapTxEntry(CTxMemPool::txiter it, indexed_modified_transaction_set &mapModifiedTx, CTxMemPool::setEntries &failedTx)
 {
     assert (it != mempool.mapTx.end());
+    if (it->GetTime() > GetTime() - nMinTxAge) return true; // txn too recent
     return mapModifiedTx.count(it) || inBlock.count(it) || failedTx.count(it);
 }
 

src/miner.h

@@ -135,6 +135,7 @@ class BlockAssembler
     bool fIncludeWitness;
     unsigned int nBlockMaxWeight;
     CFeeRate blockMinFeeRate;
+    int64_t nMinTxAge;
 
     // Information on the current status of the block
     uint64_t nBlockWeight;
@@ -153,6 +154,7 @@ class BlockAssembler
         Options();
         size_t nBlockMaxWeight;
         CFeeRate blockMinFeeRate;
+        int64_t nMinTxAge;
     };
 
     explicit BlockAssembler(const CChainParams& params);

src/net.h

@@ -723,6 +723,9 @@ class CNode
     // Used for BIP35 mempool sending
     bool fSendMempool GUARDED_BY(cs_inventory){false};
 
+    // Used for scheduling rebroadcasts
+    std::chrono::seconds m_next_rebroadcast{0};
+
     // Last time a "MEMPOOL" request was serviced.
     std::atomic<int64_t> timeLastMempoolReq{0};
 
@@ -880,11 +883,14 @@ class CNode
     void MaybeSetAddrName(const std::string& addrNameIn);
 };
 
-
-
-
-
 /** Return a timestamp in the future (in microseconds) for exponentially distributed events. */
 int64_t PoissonNextSend(int64_t now, int average_interval_seconds);
 
+/** Wrapper to return mockable type */
+inline std::chrono::seconds PoissonNextSend(std::chrono::seconds now, int average_interval_seconds)
+{
+    int64_t now_micros = (std::chrono::duration_cast<std::chrono::microseconds>(now)).count();
+    return std::chrono::duration_cast<std::chrono::seconds>(std::chrono::microseconds{PoissonNextSend(now_micros, average_interval_seconds)});
+}
+
 #endif // BITCOIN_NET_H

src/net_processing.cpp

@@ -110,6 +110,9 @@ static constexpr unsigned int INVENTORY_BROADCAST_MAX = 7 * INVENTORY_BROADCAST_
 static constexpr unsigned int AVG_FEEFILTER_BROADCAST_INTERVAL = 10 * 60;
 /** Maximum feefilter broadcast delay after significant change. */
 static constexpr unsigned int MAX_FEEFILTER_CHANGE_DELAY = 5 * 60;
+/** Average delay between rebroadcasts in seconds. */
+static const unsigned int TX_REBROADCAST_INTERVAL = 60 * 60;
+
 
 // Internal stuff
 namespace {
@@ -1533,6 +1536,12 @@ void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnm
             if (mi != mapRelay.end()) {
                 connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::TX, *mi->second));
                 push = true;
+
+                auto num = mempool.setUnbroadcastTxIDs.erase(inv.hash);
+                if (num) {
+                    LogPrint(BCLog::NET, "Removed %i from setUnbroadcastTxIDs \n", inv.hash.GetHex());
+                }
+
             } else if (pfrom->timeLastMempoolReq) {
                 auto txinfo = mempool.info(inv.hash);
                 // To protect privacy, do not answer getdata using the mempool when
@@ -3804,6 +3813,30 @@ bool PeerLogicValidation::SendMessages(CNode* pto)
                 }
             }
 
+            // Check for rebroadcasts
+            const auto current_time = GetTime<std::chrono::seconds>();
+
+            if (pto->m_next_rebroadcast < current_time) {
+                bool fFirst = (pto->m_next_rebroadcast.count() == 0);
+                pto->m_next_rebroadcast = PoissonNextSend(current_time, TX_REBROADCAST_INTERVAL);
+
+                if (!fFirst) {
+                    std::set<uint256> setRebroadcastTxs;
+                    mempool.GetRebroadcastTransactions(setRebroadcastTxs);
+
+                    for (const auto& hash : setRebroadcastTxs) {
+                        LogPrint(BCLog::NET, "Rebroadcast tx=%s peer=%d\n", hash.GetHex(), pto->GetId());
+                    }
+
+                    pto->setInventoryTxToSend.insert(setRebroadcastTxs.begin(), setRebroadcastTxs.end());
+
+                    // also ensure inclusion of wallet txns that haven't been succesfully broadcast yet
+                    // since set elements are unique, this will be a no-op if the txns are already in setInventoryTxToSend
+                    pto->setInventoryTxToSend.insert(mempool.setUnbroadcastTxIDs.begin(), mempool.setUnbroadcastTxIDs.end());
+                }
+            }
+
+
             // Time to send but the peer has requested we not relay transactions.
             if (fSendTrickle) {
                 LOCK(pto->cs_filter);

src/node/transaction.cpp

@@ -55,6 +55,9 @@ TransactionError BroadcastTransaction(const CTransactionRef tx, std::string& err
 
         // Transaction was accepted to the mempool.
 
+        // the mempool explicitly keeps track of wallet txns to ensure succesful initial broadcast
+        mempool.setUnbroadcastTxIDs.insert(hashTx);
+
         if (wait_callback) {
             // For transactions broadcast from outside the wallet, make sure
             // that the wallet has been notified of the transaction before

src/policy/policy.h

@@ -20,6 +20,8 @@ class CTxOut;
 static const unsigned int DEFAULT_BLOCK_MAX_WEIGHT = MAX_BLOCK_WEIGHT - 4000;
 /** Default for -blockmintxfee, which sets the minimum feerate for a transaction in blocks created by mining code **/
 static const unsigned int DEFAULT_BLOCK_MIN_TX_FEE = 1000;
+/** Default rebroadcast in seconds - 30 min **/
+static const int64_t REBROADCAST_MIN_TX_AGE = 30 * 60;
 /** The maximum weight for transactions we're willing to relay/mine */
 static const unsigned int MAX_STANDARD_TX_WEIGHT = 400000;
 /** The minimum non-witness size for transactions we're willing to relay/mine (1 segwit input + 1 P2WPKH output = 82 bytes) */

src/rpc/mining.cpp

@@ -218,7 +218,9 @@ static UniValue getmininginfo(const JSONRPCRequest& request)
 static UniValue prioritisetransaction(const JSONRPCRequest& request)
 {
             RPCHelpMan{"prioritisetransaction",
-                "Accepts the transaction into mined blocks at a higher (or lower) priority\n",
+                "Accepts the transaction into mined blocks at a higher (or lower) priority.\n"
+                "\nNote that prioritizing a transaction could leak privacy, through both\n"
+                "block mining and likelihood of rebroadcast.",
                 {
                     {"txid", RPCArg::Type::STR_HEX, RPCArg::Optional::NO, "The transaction id."},
                     {"dummy", RPCArg::Type::NUM, RPCArg::Optional::OMITTED_NAMED_ARG, "API-Compatibility for previous API. Must be zero or null.\n"

src/txmempool.cpp

@@ -5,17 +5,19 @@
 
 #include <txmempool.h>
 
+#include <chainparams.h>
 #include <consensus/consensus.h>
 #include <consensus/tx_verify.h>
 #include <consensus/validation.h>
-#include <validation.h>
-#include <policy/policy.h>
+#include <miner.h>
 #include <policy/fees.h>
+#include <policy/policy.h>
 #include <policy/settings.h>
 #include <reverse_iterator.h>
-#include <util/system.h>
 #include <util/moneystr.h>
+#include <util/system.h>
 #include <util/time.h>
+#include <validation.h>
 
 CTxMemPoolEntry::CTxMemPoolEntry(const CTransactionRef& _tx, const CAmount& _nFee,
                                  int64_t _nTime, unsigned int _entryHeight,
@@ -97,6 +99,29 @@ void CTxMemPool::UpdateForDescendants(txiter updateIt, cacheMap &cachedDescendan
     mapTx.modify(updateIt, update_descendant_state(modifySize, modifyFee, modifyCount));
 }
 
+void CTxMemPool::GetRebroadcastTransactions(std::set<uint256>& setRebroadcastTxs)
+{
+    // Don't rebroadcast txns during importing, reindex, or IBD to ensure we don't
+    // accidentally spam our peers with old transactions.
+    if (::ChainstateActive().IsInitialBlockDownload() || ::fImporting || ::fReindex) return;
+
+    BlockAssembler::Options options;
+    options.nBlockMaxWeight = MAX_REBROADCAST_WEIGHT;
+    options.nMinTxAge = REBROADCAST_MIN_TX_AGE;
+    CScript scriptDummy = CScript() << OP_TRUE;
+
+    // use CreateNewBlock to get set of transaction candidates
+    std::unique_ptr<CBlockTemplate> pblocktemplate = BlockAssembler(Params(), options).CreateNewBlock(scriptDummy);
+
+    LOCK(cs);
+    for (const auto& tx : pblocktemplate->block.vtx) {
+        if (mapTx.find(tx->GetHash()) == mapTx.end()) continue;
+
+        // add to rebroadcast set
+        setRebroadcastTxs.insert(tx->GetHash());
+    }
+}
+
 // vHashesToUpdate is the set of transaction hashes from a disconnected block
 // which has been re-added to the mempool.
 // for each entry, look for descendants that are outside vHashesToUpdate, and

src/txmempool.h

@@ -35,6 +35,10 @@ extern CCriticalSection cs_main;
 /** Fake height value used in Coin to signify they are only in the memory pool (since 0.8) */
 static const uint32_t MEMPOOL_HEIGHT = 0x7FFFFFFF;
 
+// we rebroadcast 3/4 of max block weight (defined in consensus.h)
+// to reduce noise due to circumstances such as miners mining priority txns
+static const unsigned int MAX_REBROADCAST_WEIGHT = 3000000;
+
 struct LockPoints
 {
     // Will be set to the blockchain height and median time past
@@ -530,6 +534,10 @@ class CTxMemPool
     const setEntries & GetMemPoolParents(txiter entry) const EXCLUSIVE_LOCKS_REQUIRED(cs);
     const setEntries & GetMemPoolChildren(txiter entry) const EXCLUSIVE_LOCKS_REQUIRED(cs);
     uint64_t CalculateDescendantMaximum(txiter entry) const EXCLUSIVE_LOCKS_REQUIRED(cs);
+
+    // track wallet transactions to ensur ethey are succesfully broadcast
+    std::set<uint256> setUnbroadcastTxIDs;
+
 private:
     typedef std::map<txiter, setEntries, CompareIteratorByHash> cacheMap;
 
@@ -614,6 +622,11 @@ class CTxMemPool
      */
     void RemoveStaged(setEntries& stage, bool updateDescendants, MemPoolRemovalReason reason) EXCLUSIVE_LOCKS_REQUIRED(cs);
 
+    /** Use CreateNewBlock with specific rebroadcast parameters to identify a set
+     * of transaction candidates & populate them in setRebroadcastTxs.
+     */
+    void GetRebroadcastTransactions(std::set<uint256>& setRebroadcastTxs);
+
     /** When adding transactions from a disconnected block back to the mempool,
      *  new mempool entries may have children in the mempool (which is generally
      *  not the case when otherwise adding transactions).

src/wallet/wallet.cpp

@@ -46,6 +46,8 @@ const std::map<uint64_t,std::string> WALLET_FLAG_CAVEATS{
 };
 
 static const size_t OUTPUT_GROUP_MAX_ENTRIES = 10;
+// frequency of resubmitting txns to mempool- 24 hours in ms
+static const int RESEND_TXS_FREQUENCY = 1000 * 60 * 60 * 24;
 
 static CCriticalSection cs_wallets;
 static std::vector<std::shared_ptr<CWallet>> vpwallets GUARDED_BY(cs_wallets);
@@ -2344,47 +2346,29 @@ bool CWalletTx::IsEquivalentTo(const CWalletTx& _tx) const
         return CTransaction(tx1) == CTransaction(tx2);
 }
 
-// Rebroadcast transactions from the wallet. We do this on a random timer
-// to slightly obfuscate which transactions come from our wallet.
-//
-// Ideally, we'd only resend transactions that we think should have been
-// mined in the most recent block. Any transaction that wasn't in the top
-// blockweight of transactions in the mempool shouldn't have been mined,
-// and so is probably just sitting in the mempool waiting to be confirmed.
-// Rebroadcasting does nothing to speed up confirmation and only damages
-// privacy.
+// Once a day, resbumit all wallet transactions to the node,
+// incase it has been dropped from your mempool.
 void CWallet::ResendWalletTransactions()
 {
     // During reindex, importing and IBD, old wallet transactions become
-    // unconfirmed. Don't resend them as that would spam other nodes.
+    // unconfirmed. Don't need to resubmit to our node.
     if (!chain().isReadyToBroadcast()) return;
 
-    // Do this infrequently and randomly to avoid giving away
-    // that these are our transactions.
-    if (GetTime() < nNextResend || !fBroadcastTransactions) return;
-    bool fFirst = (nNextResend == 0);
-    nNextResend = GetTime() + GetRand(30 * 60);
-    if (fFirst) return;
-
-    // Only do it if there's been a new block since last time
-    if (m_best_block_time < nLastResend) return;
-    nLastResend = GetTime();
+    // Do this once per day.
+    if (GetTime() < nNextResend) return;
+    nNextResend = GetTime() + RESEND_TXS_FREQUENCY;
 
     int submitted_tx_count = 0;
 
     { // locked_chain and cs_wallet scope
         auto locked_chain = chain().lock();
         LOCK(cs_wallet);
 
-        // Relay transactions
+        // Resubmit transactions
         for (std::pair<const uint256, CWalletTx>& item : mapWallet) {
             CWalletTx& wtx = item.second;
-            // Attempt to rebroadcast all txes more than 5 minutes older than
-            // the last block. SubmitMemoryPoolAndRelay() will not rebroadcast
-            // any confirmed or conflicting txs.
-            if (wtx.nTimeReceived > m_best_block_time - 5 * 60) continue;
             std::string unused_err_string;
-            if (wtx.SubmitMemoryPoolAndRelay(unused_err_string, true, *locked_chain)) ++submitted_tx_count;
+            if (wtx.SubmitMemoryPoolAndRelay(unused_err_string, false, *locked_chain)) ++submitted_tx_count;
         }
     } // locked_chain and cs_wallet
 

src/wallet/wallet.h

@@ -722,7 +722,6 @@ class CWallet final : public FillableSigningProvider, private interfaces::Chain:
     int nWalletMaxVersion GUARDED_BY(cs_wallet) = FEATURE_BASE;
 
     int64_t nNextResend = 0;
-    int64_t nLastResend = 0;
     bool fBroadcastTransactions = false;
     // Local time that the tip block was received. Used to schedule wallet rebroadcasts.
     std::atomic<int64_t> m_best_block_time {0};