Call wallet/validation notify callbacks in scheduler thread (without cs_main)

doc/developer-notes.md

@@ -907,3 +907,16 @@ A few guidelines for introducing and reviewing new RPC interfaces:
   - *Exception*: Using RPC method aliases may be appropriate in cases where a
     new RPC is replacing a deprecated RPC, to avoid both RPCs confusingly
     showing up in the command list.
+
+- Wallet RPCs call BlockUntilSyncedToCurrentChain to maintain consistency with
+  `getblockchaininfo`'s state immediately prior to the call's execution. Wallet
+  RPCs whose behavior does *not* depend on the current chainstate may omit this
+  call.
+
+  - *Rationale*: In previous versions of Bitcoin Core, the wallet was always
+    in-sync with the chainstate (by virtue of them all being updated in the
+    same cs_main lock). In order to maintain the behavior that wallet RPCs
+    return results as of at least the highest best-known block an RPC
+    client may be aware of prior to entering a wallet RPC call, we must block
+    until the wallet is caught up to the chainstate as of the RPC call's entry.
+    This also makes the API much easier for RPC clients to reason about.

src/blockassembler.cpp

@@ -70,6 +70,10 @@ bool SolveProofOfStake(CBlock* pblock, CBlockIndex* pindexPrev, CWallet* pwallet
 {
     boost::this_thread::interruption_point();
     pblock->nBits = GetNextWorkRequired(pindexPrev, pblock);
+
+    // Sync wallet before create coinstake
+    pwallet->BlockUntilSyncedToCurrentChain();
+
     CMutableTransaction txCoinStake;
     int64_t nTxNewTime = 0;
     if (!pwallet->CreateCoinStake(*pwallet, pindexPrev, pblock->nBits, txCoinStake, nTxNewTime, availableCoins)) {

src/init.cpp

@@ -318,6 +318,7 @@ void PrepareShutdown()
     // Disconnect all slots
     UnregisterAllValidationInterfaces();
     GetMainSignals().UnregisterBackgroundSignalScheduler();
+    GetMainSignals().UnregisterWithMempoolSignals(mempool);
 
 #ifndef WIN32
     try {
@@ -1264,6 +1265,7 @@ bool AppInitMain()
     threadGroup.create_thread(std::bind(&TraceThread<CScheduler::Function>, "scheduler", serviceLoop));
 
     GetMainSignals().RegisterBackgroundSignalScheduler(scheduler);
+    GetMainSignals().RegisterWithMempoolSignals(mempool);
 
     // Initialize Sapling circuit parameters
     LoadSaplingParams();

src/miner.cpp

@@ -39,10 +39,10 @@
 double dHashesPerSec = 0.0;
 int64_t nHPSTimerStart = 0;
 
-std::unique_ptr<CBlockTemplate> CreateNewBlockWithKey(CReserveKey& reservekey, CWallet* pwallet)
+std::unique_ptr<CBlockTemplate> CreateNewBlockWithKey(CReserveKey* reservekey, CWallet* pwallet)
 {
     CPubKey pubkey;
-    if (!reservekey.GetReservedKey(pubkey))
+    if (!reservekey->GetReservedKey(pubkey))
         return nullptr;
 
     const int nHeightNext = chainActive.Tip()->nHeight + 1;
@@ -169,7 +169,7 @@ void BitcoinMiner(CWallet* pwallet, bool fProofOfStake)
 
         std::unique_ptr<CBlockTemplate> pblocktemplate((fProofOfStake ?
                                                         BlockAssembler(Params(), DEFAULT_PRINTPRIORITY).CreateNewBlock(CScript(), pwallet, true, &availableCoins) :
-                                                        CreateNewBlockWithKey(*opReservekey, pwallet)));
+                                                        CreateNewBlockWithKey(opReservekey.get_ptr(), pwallet)));
         if (!pblocktemplate) continue;
         std::shared_ptr<CBlock> pblock = std::make_shared<CBlock>(pblocktemplate->block);
 

src/miner.h

@@ -27,7 +27,7 @@ static const bool DEFAULT_PRINTPRIORITY = false;
     /** Run the miner threads */
     void GenerateBitcoins(bool fGenerate, CWallet* pwallet, int nThreads);
     /** Generate a new block, without valid proof-of-work */
-    std::unique_ptr<CBlockTemplate> CreateNewBlockWithKey(CReserveKey& reservekey, CWallet* pwallet);
+    std::unique_ptr<CBlockTemplate> CreateNewBlockWithKey(CReserveKey* reservekey, CWallet* pwallet);
 
     void BitcoinMiner(CWallet* pwallet, bool fProofOfStake);
     void ThreadStakeMinter();

src/rpc/blockchain.cpp

@@ -21,6 +21,7 @@
 #include "utilmoneystr.h"
 #include "utilstrencodings.h"
 #include "hash.h"
+#include "validationinterface.h"
 #include "wallet/wallet.h"
 #include "zpiv/zpivmodule.h"
 #include "zpivchain.h"
@@ -48,6 +49,21 @@ static CUpdatedBlock latestblock;
 extern void TxToJSON(const CTransaction& tx, const uint256 hashBlock, UniValue& entry);
 void ScriptPubKeyToJSON(const CScript& scriptPubKey, UniValue& out, bool fIncludeHex);
 
+UniValue syncwithvalidationinterfacequeue(const JSONRPCRequest& request)
+{
+    if (request.fHelp || request.params.size() > 0) {
+        throw std::runtime_error(
+                "syncwithvalidationinterfacequeue\n"
+                "\nWaits for the validation interface queue to catch up on everything that was there when we entered this function.\n"
+                "\nExamples:\n"
+                + HelpExampleCli("syncwithvalidationinterfacequeue","")
+                + HelpExampleRpc("syncwithvalidationinterfacequeue","")
+        );
+    }
+    SyncWithValidationInterfaceQueue();
+    return NullUniValue;
+}
+
 double GetDifficulty(const CBlockIndex* blockindex)
 {
     // Floating point number that is a multiple of the minimum difficulty,
@@ -1448,6 +1464,7 @@ static const CRPCCommand commands[] =
     { "hidden",             "waitfornewblock",        &waitfornewblock,        true  },
     { "hidden",             "waitforblock",           &waitforblock,           true  },
     { "hidden",             "waitforblockheight",     &waitforblockheight,     true  },
+    { "hidden",             "syncwithvalidationinterfacequeue", &syncwithvalidationinterfacequeue, true },
 
 
 };

src/rpc/mining.cpp

@@ -61,14 +61,17 @@ UniValue generate(const JSONRPCRequest& request)
 
     const Consensus::Params& consensus = Params().GetConsensus();
     bool fPoS = consensus.NetworkUpgradeActive(nHeight + 1, Consensus::UPGRADE_POS);
+    std::unique_ptr<CReserveKey> reservekey;
 
     if (fPoS) {
         // If we are in PoS, wallet must be unlocked.
         EnsureWalletIsUnlocked();
+    } else {
+        // Coinbase key
+        reservekey = MakeUnique<CReserveKey>(pwalletMain);
     }
 
     UniValue blockHashes(UniValue::VARR);
-    CReserveKey reservekey(pwalletMain);
     unsigned int nExtraNonce = 0;
 
     while (nHeight < nHeightEnd && !ShutdownRequested()) {
@@ -81,7 +84,7 @@ UniValue generate(const JSONRPCRequest& request)
 
         std::unique_ptr<CBlockTemplate> pblocktemplate((fPoS ?
                                                         BlockAssembler(Params(), DEFAULT_PRINTPRIORITY).CreateNewBlock(CScript(), pwalletMain, true, &availableCoins) :
-                                                        CreateNewBlockWithKey(reservekey, pwalletMain)));
+                                                        CreateNewBlockWithKey(reservekey.get(), pwalletMain)));
         if (!pblocktemplate.get()) break;
         std::shared_ptr<CBlock> pblock = std::make_shared<CBlock>(pblocktemplate->block);
 
@@ -114,6 +117,12 @@ UniValue generate(const JSONRPCRequest& request)
     if (nGenerated == 0 || (!fPoS && nGenerated < nGenerate))
         throw JSONRPCError(RPC_INTERNAL_ERROR, "Couldn't create new blocks");
 
+    // mark key as used, only for PoW coinbases
+    if (reservekey) {
+        // Remove key from key pool
+        reservekey->KeepKey();
+    }
+
     return blockHashes;
 }
 #endif // ENABLE_WALLET

src/rpc/rawtransaction.cpp

@@ -9,6 +9,7 @@
 #include "core_io.h"
 #include "init.h"
 #include "keystore.h"
+#include "validationinterface.h"
 #include "net.h"
 #include "policy/policy.h"
 #include "primitives/transaction.h"
@@ -25,6 +26,7 @@
 #include "wallet/wallet.h"
 #endif
 
+#include <future>
 #include <stdint.h>
 
 #include <univalue.h>
@@ -518,6 +520,10 @@ UniValue fundrawtransaction(const JSONRPCRequest& request)
     if (!pwalletMain)
         throw std::runtime_error("wallet not initialized");
 
+    // Make sure the results are valid at least up to the most recent block
+    // the user could have gotten from another RPC command prior to now
+    pwalletMain->BlockUntilSyncedToCurrentChain();
+
     RPCTypeCheck(request.params, {UniValue::VSTR});
 
     CTxDestination changeAddress = CNoDestination();
@@ -899,6 +905,8 @@ UniValue sendrawtransaction(const JSONRPCRequest& request)
             "\nSend the transaction (signed hex)\n" + HelpExampleCli("sendrawtransaction", "\"signedhex\"") +
             "\nAs a json rpc call\n" + HelpExampleRpc("sendrawtransaction", "\"signedhex\""));
 
+    std::promise<void> promise;
+
     RPCTypeCheck(request.params, {UniValue::VSTR, UniValue::VBOOL});
 
     // parse hex string from parameter
@@ -911,7 +919,8 @@ UniValue sendrawtransaction(const JSONRPCRequest& request)
     if (request.params.size() > 1)
         fOverrideFees = request.params[1].get_bool();
 
-    AssertLockNotHeld(cs_main);
+    { // cs_main scope
+    LOCK(cs_main);
     CCoinsViewCache& view = *pcoinsTip;
     bool fHaveChain = false;
     for (size_t o = 0; !fHaveChain && o < mtx.vout.size(); o++) {
@@ -923,7 +932,6 @@ UniValue sendrawtransaction(const JSONRPCRequest& request)
         CValidationState state;
         bool fMissingInputs;
         {
-            LOCK(cs_main);
             if (!AcceptToMemoryPool(mempool, state, MakeTransactionRef(std::move(mtx)), true, &fMissingInputs, false, !fOverrideFees)) {
                 if (state.IsInvalid()) {
                     throw JSONRPCError(RPC_TRANSACTION_REJECTED, strprintf("%i: %s", state.GetRejectCode(), state.GetRejectReason()));
@@ -933,11 +941,25 @@ UniValue sendrawtransaction(const JSONRPCRequest& request)
                     }
                     throw JSONRPCError(RPC_TRANSACTION_ERROR, state.GetRejectReason());
                 }
+            } else {
+                // If wallet is enabled, ensure that the wallet has been made aware
+                // of the new transaction prior to returning. This prevents a race
+                // where a user might call sendrawtransaction with a transaction
+                // to/from their wallet, immediately call some wallet RPC, and get
+                // a stale result because callbacks have not yet been processed.
+                CallFunctionInValidationInterfaceQueue([&promise] {
+                    promise.set_value();
+                });
             }
         }
     } else if (fHaveChain) {
         throw JSONRPCError(RPC_TRANSACTION_ALREADY_IN_CHAIN, "transaction already in block chain");
     }
+
+    } // cs_main
+
+    promise.get_future().wait();
+
     if(!g_connman)
         throw JSONRPCError(RPC_CLIENT_P2P_DISABLED, "Error: Peer-to-peer functionality missing or disabled");
 
@@ -946,6 +968,7 @@ UniValue sendrawtransaction(const JSONRPCRequest& request)
     {
         pnode->PushInventory(inv);
     });
+
     return hashTx.GetHex();
 }
 

src/sapling/saplingscriptpubkeyman.cpp

@@ -760,7 +760,7 @@ CAmount SaplingScriptPubKeyMan::GetShieldedChange(const CWalletTx& wtx) const
 
 bool SaplingScriptPubKeyMan::IsNoteSaplingChange(const SaplingOutPoint& op, libzcash::SaplingPaymentAddress address) const
 {
-    LOCK(wallet->cs_KeyStore);
+    LOCK2(wallet->cs_wallet, wallet->cs_KeyStore);
     std::set<libzcash::PaymentAddress> shieldedAddresses = {address};
     std::set<std::pair<libzcash::PaymentAddress, uint256>> nullifierSet = GetNullifiersForAddresses(shieldedAddresses);
     return IsNoteSaplingChange(nullifierSet, address, op);

src/scheduler.cpp

@@ -186,3 +186,8 @@ void SingleThreadedSchedulerClient::EmptyQueue() {
         should_continue = !m_callbacks_pending.empty();
     }
 }
+
+size_t SingleThreadedSchedulerClient::CallbacksPending() {
+    LOCK(m_cs_callbacks_pending);
+    return m_callbacks_pending.size();
+}

src/scheduler.h

@@ -109,6 +109,8 @@ class SingleThreadedSchedulerClient {
     // Processes all remaining queue members on the calling thread, blocking until queue is empty
     // Must be called after the CScheduler has no remaining processing threads!
     void EmptyQueue();
+
+    size_t CallbacksPending();
 };
 
 #endif

src/test/librust/sapling_rpc_wallet_tests.cpp

@@ -87,7 +87,11 @@ BOOST_AUTO_TEST_CASE(rpc_wallet_getbalance)
 {
     SelectParams(CBaseChainParams::TESTNET);
 
-    LOCK2(cs_main, pwalletMain->cs_wallet);
+    {
+        LOCK(pwalletMain->cs_wallet);
+        pwalletMain->SetMinVersion(FEATURE_SAPLING);
+        pwalletMain->SetupSPKM(false);
+    }
 
     BOOST_CHECK_THROW(CallRPC("getshieldbalance too many args"), std::runtime_error);
     BOOST_CHECK_THROW(CallRPC("getshieldbalance invalidaddress"), std::runtime_error);
@@ -249,7 +253,11 @@ BOOST_AUTO_TEST_CASE(rpc_shieldsendmany_parameters)
 {
     SelectParams(CBaseChainParams::TESTNET);
 
-    LOCK2(cs_main, pwalletMain->cs_wallet);
+    {
+        LOCK(pwalletMain->cs_wallet);
+        pwalletMain->SetMinVersion(FEATURE_SAPLING);
+        pwalletMain->SetupSPKM(false);
+    }
 
     BOOST_CHECK_THROW(CallRPC("shieldsendmany"), std::runtime_error);
     BOOST_CHECK_THROW(CallRPC("shieldsendmany toofewargs"), std::runtime_error);
@@ -299,7 +307,6 @@ BOOST_AUTO_TEST_CASE(rpc_shieldsendmany_parameters)
     std::vector<char> v (2 * (ZC_MEMO_SIZE+1));     // x2 for hexadecimal string format
     std::fill(v.begin(),v.end(), 'A');
     std::string badmemo(v.begin(), v.end());
-    pwalletMain->SetupSPKM(false);
     auto pa = pwalletMain->GenerateNewSaplingZKey();
     std::string zaddr1 = KeyIO::EncodePaymentAddress(pa);
     BOOST_CHECK_THROW(CallRPC(std::string("shieldsendmany yBYhwgzufrZ6F5VVuK9nEChENArq934mqC ")
@@ -543,8 +550,10 @@ BOOST_AUTO_TEST_CASE(rpc_listshieldunspent_parameters)
 {
     SelectParams(CBaseChainParams::TESTNET);
 
-    LOCK2(cs_main, pwalletMain->cs_wallet);
-    pwalletMain->SetupSPKM(false);
+    {
+        LOCK(pwalletMain->cs_wallet);
+        pwalletMain->SetupSPKM(false);
+    }
 
     UniValue retValue;
 

src/test/test_pivx.cpp

@@ -54,10 +54,15 @@ TestingSetup::TestingSetup()
         fs::create_directories(pathTemp);
         gArgs.ForceSetArg("-datadir", pathTemp.string());
 
+        // Start the lightweight task scheduler thread
+        CScheduler::Function serviceLoop = std::bind(&CScheduler::serviceQueue, &scheduler);
+        threadGroup.create_thread(std::bind(&TraceThread<CScheduler::Function>, "scheduler", serviceLoop));
+
         // Note that because we don't bother running a scheduler thread here,
         // callbacks via CValidationInterface are unreliable, but that's OK,
         // our unit tests aren't testing multiple parts of the code at once.
         GetMainSignals().RegisterBackgroundSignalScheduler(scheduler);
+        GetMainSignals().RegisterWithMempoolSignals(mempool);
 
         // Ideally we'd move all the RPC tests to the functional testing framework
         // instead of unit tests, but for now we need these here.
@@ -87,7 +92,9 @@ TestingSetup::~TestingSetup()
         threadGroup.interrupt_all();
         threadGroup.join_all();
         GetMainSignals().FlushBackgroundCallbacks();
+        UnregisterAllValidationInterfaces();
         GetMainSignals().UnregisterBackgroundSignalScheduler();
+        GetMainSignals().UnregisterWithMempoolSignals(mempool);
         UnloadBlockIndex();
         delete pcoinsTip;
         delete pcoinsdbview;

src/txmempool.h

@@ -535,6 +535,9 @@ class CTxMemPool
     // to track size/count of descendant transactions.  First version of
     // addUnchecked can be used to have it call CalculateMemPoolAncestors(), and
     // then invoke the second version.
+    // Note that addUnchecked is ONLY called from ATMP outside of tests
+    // and any other callers may break wallet's in-mempool tracking (due to
+    // lack of CValidationInterface::TransactionAddedToMempool callbacks).
     bool addUnchecked(const uint256& hash, const CTxMemPoolEntry& entry, bool fCurrentEstimate = true);
     bool addUnchecked(const uint256& hash, const CTxMemPoolEntry &entry, setEntries &setAncestors, bool fCurrentEstimate = true);