Merge bitcoin/bitcoin#25487: [kernel 3b/n] Decouple {Dump,Load}Mempool from ArgsManager

cb3e9a1 Move {Load,Dump}Mempool to kernel namespace (Carl Dong)
aa30676 Move DEFAULT_PERSIST_MEMPOOL out of libbitcoinkernel (Carl Dong)
06b88ff LoadMempool: Pass in load_path, stop using gArgs (Carl Dong)
b857ac6 test/fuzz: Invoke LoadMempool via CChainState (Carl Dong)
b326725 Move FopenFn to fsbridge namespace (Carl Dong)
ae1e8e3 mempool: Use NodeClock+friends for LoadMempool (Carl Dong)
f9e8e57 mempool: Improve comments for [GS]etLoadTried (Carl Dong)
813962d scripted-diff: Rename m_is_loaded -> m_load_tried (Carl Dong)
413f4bb DumpMempool: Pass in dump_path, stop using gArgs (Carl Dong)
bd44078 DumpMempool: Use std::chrono instead of weird int64_t arthmetics (Carl Dong)
c84390b test/mempool_persist: Test manual savemempool when -persistmempool=0 (Carl Dong)

Pull request description:

  This is part of the `libbitcoinkernel` project: #24303, https://github.com/bitcoin/bitcoin/projects/18

  -----

  This PR moves `{Dump,Load}Mempool` into its own `kernel/mempool_persist` module and introduces `ArgsManager` `node::` helpers in `node/mempool_persist_args`to remove the scattered calls to `GetBoolArg("-persistmempool", DEFAULT_PERSIST_MEMPOOL)`.

  More context can be gleaned from the commit messages.

  -----

  One thing I was reflecting on as I wrote this was that in the long run, I think we should probably invert the validation <-> mempool relationship. Instead of mempool not depending on validation, it might make more sense to have validation not depend on mempool. Not super urgent since `libbitcoinkernel` will include both validation and mempool, but perhaps something for the future.

ACKs for top commit:
  glozow:
    re ACK cb3e9a1 via `git range-diff 7ae032e...cb3e9a1`
  MarcoFalke:
    ACK cb3e9a1 🔒
  ryanofsky:
    Code review ACK cb3e9a1

Tree-SHA512: 979d7237c3abb5a1dd9b5ad3dbf3b954f906a6d8320ed7b923557f41a4472deccae3e8a6bca0018c8e7a3c4a93afecc502acd1e26756f2054f157f1c0edd939d

Author: glozow

ci/test/06_script_b.sh

@@ -41,6 +41,7 @@ if [ "${RUN_TIDY}" = "true" ]; then
   CI_EXEC "python3 ${DIR_IWYU}/include-what-you-use/iwyu_tool.py"\
           " src/compat"\
           " src/init"\
+          " src/kernel/mempool_persist.cpp"\
           " src/policy/feerate.cpp"\
           " src/policy/packages.cpp"\
           " src/policy/settings.cpp"\

contrib/devtools/iwyu/bitcoin.core.imp

@@ -3,4 +3,5 @@
   { include: [ "<bits/termios-c_lflag.h>", private, "<termios.h>", public ] },
   { include: [ "<bits/termios-struct.h>", private, "<termios.h>", public ] },
   { include: [ "<bits/termios-tcflow.h>", private, "<termios.h>", public ] },
+  { include: [ "<bits/chrono.h>", private, "<chrono>", public ] },
 ]

src/Makefile.am

@@ -177,6 +177,7 @@ BITCOIN_CORE_H = \
   kernel/context.h \
   kernel/mempool_limits.h \
   kernel/mempool_options.h \
+  kernel/mempool_persist.h \
   key.h \
   key_io.h \
   logging.h \
@@ -198,6 +199,7 @@ BITCOIN_CORE_H = \
   node/chainstate.h \
   node/coin.h \
   node/context.h \
+  node/mempool_persist_args.h \
   node/miner.h \
   node/minisketchwrapper.h \
   node/psbt.h \
@@ -366,6 +368,7 @@ libbitcoin_node_a_SOURCES = \
   kernel/checks.cpp \
   kernel/coinstats.cpp \
   kernel/context.cpp \
+  kernel/mempool_persist.cpp \
   mapport.cpp \
   mempool_args.cpp \
   net.cpp \
@@ -379,6 +382,7 @@ libbitcoin_node_a_SOURCES = \
   node/context.cpp \
   node/eviction.cpp \
   node/interfaces.cpp \
+  node/mempool_persist_args.cpp \
   node/miner.cpp \
   node/minisketchwrapper.cpp \
   node/psbt.cpp \
@@ -881,6 +885,7 @@ libbitcoinkernel_la_SOURCES = \
   kernel/checks.cpp \
   kernel/coinstats.cpp \
   kernel/context.cpp \
+  kernel/mempool_persist.cpp \
   key.cpp \
   logging.cpp \
   node/blockstorage.cpp \

src/Makefile.test_fuzz.include

@@ -10,6 +10,7 @@ EXTRA_LIBRARIES += \
 TEST_FUZZ_H = \
     test/fuzz/fuzz.h \
     test/fuzz/FuzzedDataProvider.h \
+    test/fuzz/mempool_utils.h \
     test/fuzz/util.h
 
 libtest_fuzz_a_CPPFLAGS = $(AM_CPPFLAGS) $(BITCOIN_INCLUDES) $(MINIUPNPC_CPPFLAGS) $(NATPMP_CPPFLAGS) $(EVENT_CFLAGS) $(EVENT_PTHREADS_CFLAGS)

src/fs.h

@@ -9,6 +9,7 @@
 
 #include <cstdio>
 #include <filesystem>
+#include <functional>
 #include <iomanip>
 #include <ios>
 #include <ostream>
@@ -199,6 +200,7 @@ bool create_directories(const std::filesystem::path& p, std::error_code& ec) = d
 
 /** Bridge operations to C stdio */
 namespace fsbridge {
+    using FopenFn = std::function<FILE*(const fs::path&, const char*)>;
     FILE *fopen(const fs::path& p, const char *mode);
 
     /**

src/init.cpp

@@ -10,6 +10,7 @@
 #include <init.h>
 
 #include <kernel/checks.h>
+#include <kernel/mempool_persist.h>
 
 #include <addrman.h>
 #include <banman.h>
@@ -41,6 +42,7 @@
 #include <node/chainstate.h>
 #include <node/context.h>
 #include <node/interface_ui.h>
+#include <node/mempool_persist_args.h>
 #include <node/miner.h>
 #include <policy/feerate.h>
 #include <policy/fees.h>
@@ -102,14 +104,19 @@
 #include <zmq/zmqrpc.h>
 #endif
 
+using kernel::DumpMempool;
+
 using node::CacheSizes;
 using node::CalculateCacheSizes;
 using node::ChainstateLoadVerifyError;
 using node::ChainstateLoadingError;
 using node::CleanupBlockRevFiles;
+using node::DEFAULT_PERSIST_MEMPOOL;
 using node::DEFAULT_PRINTPRIORITY;
 using node::DEFAULT_STOPAFTERBLOCKIMPORT;
 using node::LoadChainstate;
+using node::MempoolPath;
+using node::ShouldPersistMempool;
 using node::NodeContext;
 using node::ThreadImport;
 using node::VerifyLoadedChainstate;
@@ -245,8 +252,8 @@ void Shutdown(NodeContext& node)
     node.addrman.reset();
     node.netgroupman.reset();
 
-    if (node.mempool && node.mempool->IsLoaded() && node.args->GetBoolArg("-persistmempool", DEFAULT_PERSIST_MEMPOOL)) {
-        DumpMempool(*node.mempool);
+    if (node.mempool && node.mempool->GetLoadTried() && ShouldPersistMempool(*node.args)) {
+        DumpMempool(*node.mempool, MempoolPath(*node.args));
     }
 
     // Drop transactions we were still watching, and record fee estimations.
@@ -1669,7 +1676,7 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     }
 
     chainman.m_load_block = std::thread(&util::TraceThread, "loadblk", [=, &chainman, &args] {
-        ThreadImport(chainman, vImportFiles, args);
+        ThreadImport(chainman, vImportFiles, args, ShouldPersistMempool(args) ? MempoolPath(args) : fs::path{});
     });
 
     // Wait for genesis block to be processed

src/kernel/mempool_persist.cpp

@@ -0,0 +1,189 @@
+// Copyright (c) 2022 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <kernel/mempool_persist.h>
+
+#include <clientversion.h>
+#include <consensus/amount.h>
+#include <fs.h>
+#include <logging.h>
+#include <primitives/transaction.h>
+#include <serialize.h>
+#include <shutdown.h>
+#include <streams.h>
+#include <sync.h>
+#include <txmempool.h>
+#include <uint256.h>
+#include <util/system.h>
+#include <util/time.h>
+#include <validation.h>
+
+#include <chrono>
+#include <cstdint>
+#include <cstdio>
+#include <exception>
+#include <functional>
+#include <map>
+#include <memory>
+#include <set>
+#include <stdexcept>
+#include <utility>
+#include <vector>
+
+using fsbridge::FopenFn;
+
+namespace kernel {
+
+static const uint64_t MEMPOOL_DUMP_VERSION = 1;
+
+bool LoadMempool(CTxMemPool& pool, const fs::path& load_path, CChainState& active_chainstate, FopenFn mockable_fopen_function)
+{
+    if (load_path.empty()) return false;
+
+    FILE* filestr{mockable_fopen_function(load_path, "rb")};
+    CAutoFile file(filestr, SER_DISK, CLIENT_VERSION);
+    if (file.IsNull()) {
+        LogPrintf("Failed to open mempool file from disk. Continuing anyway.\n");
+        return false;
+    }
+
+    int64_t count = 0;
+    int64_t expired = 0;
+    int64_t failed = 0;
+    int64_t already_there = 0;
+    int64_t unbroadcast = 0;
+    auto now = NodeClock::now();
+
+    try {
+        uint64_t version;
+        file >> version;
+        if (version != MEMPOOL_DUMP_VERSION) {
+            return false;
+        }
+        uint64_t num;
+        file >> num;
+        while (num) {
+            --num;
+            CTransactionRef tx;
+            int64_t nTime;
+            int64_t nFeeDelta;
+            file >> tx;
+            file >> nTime;
+            file >> nFeeDelta;
+
+            CAmount amountdelta = nFeeDelta;
+            if (amountdelta) {
+                pool.PrioritiseTransaction(tx->GetHash(), amountdelta);
+            }
+            if (nTime > TicksSinceEpoch<std::chrono::seconds>(now - pool.m_expiry)) {
+                LOCK(cs_main);
+                const auto& accepted = AcceptToMemoryPool(active_chainstate, tx, nTime, /*bypass_limits=*/false, /*test_accept=*/false);
+                if (accepted.m_result_type == MempoolAcceptResult::ResultType::VALID) {
+                    ++count;
+                } else {
+                    // mempool may contain the transaction already, e.g. from
+                    // wallet(s) having loaded it while we were processing
+                    // mempool transactions; consider these as valid, instead of
+                    // failed, but mark them as 'already there'
+                    if (pool.exists(GenTxid::Txid(tx->GetHash()))) {
+                        ++already_there;
+                    } else {
+                        ++failed;
+                    }
+                }
+            } else {
+                ++expired;
+            }
+            if (ShutdownRequested())
+                return false;
+        }
+        std::map<uint256, CAmount> mapDeltas;
+        file >> mapDeltas;
+
+        for (const auto& i : mapDeltas) {
+            pool.PrioritiseTransaction(i.first, i.second);
+        }
+
+        std::set<uint256> unbroadcast_txids;
+        file >> unbroadcast_txids;
+        unbroadcast = unbroadcast_txids.size();
+        for (const auto& txid : unbroadcast_txids) {
+            // Ensure transactions were accepted to mempool then add to
+            // unbroadcast set.
+            if (pool.get(txid) != nullptr) pool.AddUnbroadcastTx(txid);
+        }
+    } catch (const std::exception& e) {
+        LogPrintf("Failed to deserialize mempool data on disk: %s. Continuing anyway.\n", e.what());
+        return false;
+    }
+
+    LogPrintf("Imported mempool transactions from disk: %i succeeded, %i failed, %i expired, %i already there, %i waiting for initial broadcast\n", count, failed, expired, already_there, unbroadcast);
+    return true;
+}
+
+bool DumpMempool(const CTxMemPool& pool, const fs::path& dump_path, FopenFn mockable_fopen_function, bool skip_file_commit)
+{
+    auto start = SteadyClock::now();
+
+    std::map<uint256, CAmount> mapDeltas;
+    std::vector<TxMempoolInfo> vinfo;
+    std::set<uint256> unbroadcast_txids;
+
+    static Mutex dump_mutex;
+    LOCK(dump_mutex);
+
+    {
+        LOCK(pool.cs);
+        for (const auto &i : pool.mapDeltas) {
+            mapDeltas[i.first] = i.second;
+        }
+        vinfo = pool.infoAll();
+        unbroadcast_txids = pool.GetUnbroadcastTxs();
+    }
+
+    auto mid = SteadyClock::now();
+
+    try {
+        FILE* filestr{mockable_fopen_function(dump_path + ".new", "wb")};
+        if (!filestr) {
+            return false;
+        }
+
+        CAutoFile file(filestr, SER_DISK, CLIENT_VERSION);
+
+        uint64_t version = MEMPOOL_DUMP_VERSION;
+        file << version;
+
+        file << (uint64_t)vinfo.size();
+        for (const auto& i : vinfo) {
+            file << *(i.tx);
+            file << int64_t{count_seconds(i.m_time)};
+            file << int64_t{i.nFeeDelta};
+            mapDeltas.erase(i.tx->GetHash());
+        }
+
+        file << mapDeltas;
+
+        LogPrintf("Writing %d unbroadcast transactions to disk.\n", unbroadcast_txids.size());
+        file << unbroadcast_txids;
+
+        if (!skip_file_commit && !FileCommit(file.Get()))
+            throw std::runtime_error("FileCommit failed");
+        file.fclose();
+        if (!RenameOver(dump_path + ".new", dump_path)) {
+            throw std::runtime_error("Rename failed");
+        }
+        auto last = SteadyClock::now();
+
+        LogPrintf("Dumped mempool: %gs to copy, %gs to dump\n",
+                  Ticks<SecondsDouble>(mid - start),
+                  Ticks<SecondsDouble>(last - mid));
+    } catch (const std::exception& e) {
+        LogPrintf("Failed to dump mempool: %s. Continuing anyway.\n", e.what());
+        return false;
+    }
+    return true;
+}
+
+} // namespace kernel

src/kernel/mempool_persist.h

@@ -0,0 +1,28 @@
+// Copyright (c) 2022 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_KERNEL_MEMPOOL_PERSIST_H
+#define BITCOIN_KERNEL_MEMPOOL_PERSIST_H
+
+#include <fs.h>
+
+class CChainState;
+class CTxMemPool;
+
+namespace kernel {
+
+/** Dump the mempool to disk. */
+bool DumpMempool(const CTxMemPool& pool, const fs::path& dump_path,
+                 fsbridge::FopenFn mockable_fopen_function = fsbridge::fopen,
+                 bool skip_file_commit = false);
+
+/** Load the mempool from disk. */
+bool LoadMempool(CTxMemPool& pool, const fs::path& load_path,
+                 CChainState& active_chainstate,
+                 fsbridge::FopenFn mockable_fopen_function = fsbridge::fopen);
+
+} // namespace kernel
+
+
+#endif // BITCOIN_KERNEL_MEMPOOL_PERSIST_H

src/node/blockstorage.cpp

@@ -823,7 +823,7 @@ struct CImportingNow {
     }
 };
 
-void ThreadImport(ChainstateManager& chainman, std::vector<fs::path> vImportFiles, const ArgsManager& args)
+void ThreadImport(ChainstateManager& chainman, std::vector<fs::path> vImportFiles, const ArgsManager& args, const fs::path& mempool_path)
 {
     SetSyscallSandboxPolicy(SyscallSandboxPolicy::INITIALIZATION_LOAD_BLOCKS);
     ScheduleBatchPriority();
@@ -893,6 +893,6 @@ void ThreadImport(ChainstateManager& chainman, std::vector<fs::path> vImportFile
             return;
         }
     } // End scope of CImportingNow
-    chainman.ActiveChainstate().LoadMempool(args);
+    chainman.ActiveChainstate().LoadMempool(mempool_path);
 }
 } // namespace node

src/node/blockstorage.h

@@ -211,7 +211,7 @@ bool ReadRawBlockFromDisk(std::vector<uint8_t>& block, const FlatFilePos& pos, c
 
 bool UndoReadFromDisk(CBlockUndo& blockundo, const CBlockIndex* pindex);
 
-void ThreadImport(ChainstateManager& chainman, std::vector<fs::path> vImportFiles, const ArgsManager& args);
+void ThreadImport(ChainstateManager& chainman, std::vector<fs::path> vImportFiles, const ArgsManager& args, const fs::path& mempool_path);
 } // namespace node
 
 #endif // BITCOIN_NODE_BLOCKSTORAGE_H