Merge 0f68c47e931de05200adeae639bcee50ea3c171d into 5f4422d68dc3530c353af1f87499de1c864b60ad

3301d2cbe8c3b76c97285d75fa59637cb6952d0b qa: Wait for txindex to avoid race condition (Hodlinator)
9bfb0d75ba10591cc6c9620f9fd1ecc0e55e7a48 qa: Remove unnecessary -txindex args (Hodlinator)
7ac281c19cd3d11f316dbbb3308eabf1ad4f26d6 qa: Add missing coverage of corrupt indexes (Hodlinator)

Pull request description:

  - Add synchronization in 3 places where if the Transaction Index happens to be slow, we get rare test failures when querying it for transactions (one such case experienced on Windows, prompting investigation).
  - Remove unnecessary TxIndex initialization in some tests.
  - Add some test coverage where TxIndex aspect could be tested in feature_init.py.

ACKs for top commit:
  fjahr:
    re-ACK 3301d2cbe8c3b76c97285d75fa59637cb6952d0b
  mzumsande:
    Code Review ACK 3301d2cbe8c3b76c97285d75fa59637cb6952d0b
  furszy:
    Code review ACK 3301d2cbe8c3b76c97285d75fa59637cb6952d0b
  Prabhat1308:
    Concept ACK [`3301d2c`](3301d2cbe8)

Tree-SHA512: 7c2019e38455f344856aaf6b381faafbd88d53dc88d13309deb718c1dcfbee4ccca7c7f1b66917395503a6f94c3b216a007ad432cc8b93d0309db9805f38d602

cmake/bitcoin-build-config.h.in

@@ -105,6 +105,12 @@
 /* Define this symbol if the BSD sysctl(KERN_ARND) is available */
 #cmakedefine HAVE_SYSCTL_ARND 1
 
+/* Define this symbol if you have clock_gettime() */
+#cmakedefine HAVE_CLOCK_GETTIME 1
+
+/* Define this symbol if you have GetThreadTimes() */
+#cmakedefine HAVE_GETTHREADTIMES 1
+
 /* Define to 1 if std::system or ::wsystem is available. */
 #cmakedefine HAVE_SYSTEM 1
 

cmake/introspection.cmake

@@ -159,6 +159,40 @@ check_cxx_source_compiles("
   " HAVE_SYSCTL_ARND
 )
 
+# Check for clock_gettime() (POSIX.1b).
+check_cxx_source_compiles("
+  #include <time.h>
+
+  int main(int, char**)
+  {
+    timespec now;
+    clock_gettime(CLOCK_THREAD_CPUTIME_ID, &now);
+    (void)now.tv_sec;
+    (void)now.tv_nsec;
+    return 0;
+  }
+  " HAVE_CLOCK_GETTIME
+)
+
+# Check for GetThreadTimes() (Windows).
+check_cxx_source_compiles("
+  #include <windows.h>
+  #include <winnt.h>
+
+  #include <processthreadsapi.h>
+
+  int main(int, char**)
+  {
+    FILETIME creation;
+    FILETIME exit;
+    FILETIME kernel;
+    FILETIME user;
+    (void)GetThreadTimes(GetCurrentThread(), &creation, &exit, &kernel, &user);
+    return 0;
+  }
+  " HAVE_GETTHREADTIMES
+)
+
 if(NOT MSVC)
   include(CheckSourceCompilesWithFlags)
 

doc/release-notes-31672.md

@@ -0,0 +1,8 @@
+RPC
+---
+
+A new field `cpu_load` has been added to the `getpeerinfo` RPC output.
+It shows the CPU time (user + system) spent processing messages from the
+given peer and crafting messages for it expressed in per milles (‰) of
+the duration of the connection. The field is optional and will be omitted
+on platforms that do not support this or if still not measured. (#31672)

src/net.cpp

@@ -660,6 +660,8 @@ void CNode::CopyStats(CNodeStats& stats)
     stats.addrLocal = addrLocalUnlocked.IsValid() ? addrLocalUnlocked.ToStringAddrPort() : "";
 
     X(m_conn_type);
+
+    X(m_cpu_time);
 }
 #undef X
 
@@ -3047,6 +3049,8 @@ void CConnman::ThreadMessageHandler()
                 if (pnode->fDisconnect)
                     continue;
 
+                CpuTimer timer{[&pnode](std::chrono::nanoseconds elapsed) { pnode->m_cpu_time += elapsed; }};
+
                 // Receive messages
                 bool fMoreNodeWork = m_msgproc->ProcessMessages(pnode, flagInterruptMsgProc);
                 fMoreWork |= (fMoreNodeWork && !pnode->fPauseSend);

src/net.h

@@ -31,6 +31,7 @@
 #include <util/check.h>
 #include <util/sock.h>
 #include <util/threadinterrupt.h>
+#include <util/time.h>
 
 #include <atomic>
 #include <condition_variable>
@@ -220,6 +221,8 @@ public:
     TransportProtocolType m_transport_type;
     /** BIP324 session id string in hex, if any. */
     std::string m_session_id;
+    /** CPU time spent processing messages from this node and crafting messages for it. */
+    std::chrono::nanoseconds m_cpu_time;
 };
 
 
@@ -969,6 +972,9 @@ public:
         m_min_ping_time = std::min(m_min_ping_time.load(), ping_time);
     }
 
+    /** CPU time spent processing messages from this node and crafting messages for it. */
+    std::atomic<std::chrono::nanoseconds> m_cpu_time;
+
 private:
     const NodeId id;
     const uint64_t nLocalHostNonce;

src/rpc/net.cpp

@@ -143,6 +143,11 @@ static RPCHelpMan getpeerinfo()
                     {RPCResult::Type::NUM_TIME, "last_block", "The " + UNIX_EPOCH_TIME + " of the last block received from this peer"},
                     {RPCResult::Type::NUM, "bytessent", "The total bytes sent"},
                     {RPCResult::Type::NUM, "bytesrecv", "The total bytes received"},
+                    {RPCResult::Type::NUM, "cpu_load", /*optional=*/true,
+                        "The CPU time (user + system) spent processing messages from this peer "
+                        "and crafting messages for it expressed in per milles (‰) of the "
+                        "duration of the connection. Will be omitted on platforms that do not "
+                        "support this or if still not measured."},
                     {RPCResult::Type::NUM_TIME, "conntime", "The " + UNIX_EPOCH_TIME + " of the connection"},
                     {RPCResult::Type::NUM, "timeoffset", "The time offset in seconds"},
                     {RPCResult::Type::NUM, "pingtime", /*optional=*/true, "The last ping time in milliseconds (ms), if any"},
@@ -205,6 +210,8 @@ static RPCHelpMan getpeerinfo()
 
     UniValue ret(UniValue::VARR);
 
+    const auto now{GetTime<std::chrono::seconds>()};
+
     for (const CNodeStats& stats : vstats) {
         UniValue obj(UniValue::VOBJ);
         CNodeStateStats statestats;
@@ -239,6 +246,9 @@ static RPCHelpMan getpeerinfo()
         obj.pushKV("last_block", count_seconds(stats.m_last_block_time));
         obj.pushKV("bytessent", stats.nSendBytes);
         obj.pushKV("bytesrecv", stats.nRecvBytes);
+        if (stats.m_cpu_time > 0s && now > stats.m_connected) {
+            obj.pushKV("cpu_load", /* ‰ */1000.0 * stats.m_cpu_time / (now - stats.m_connected));
+        }
         obj.pushKV("conntime", count_seconds(stats.m_connected));
         obj.pushKV("timeoffset", Ticks<std::chrono::seconds>(statestats.time_offset));
         if (stats.m_last_ping_time > 0us) {

src/util/time.cpp

@@ -3,6 +3,8 @@
 // Distributed under the MIT software license, see the accompanying
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
+#include <bitcoin-build-config.h> // IWYU pragma: keep
+
 #include <util/time.h>
 
 #include <compat/compat.h>
@@ -17,6 +19,16 @@
 #include <string_view>
 #include <thread>
 
+#ifdef HAVE_CLOCK_GETTIME
+#include <time.h>
+#elif defined(HAVE_GETTHREADTIMES)
+#include <windows.h>
+#include <winnt.h>
+
+#include <processthreadsapi.h>
+#endif
+
+
 void UninterruptibleSleep(const std::chrono::microseconds& n) { std::this_thread::sleep_for(n); }
 
 static std::atomic<std::chrono::seconds> g_mock_time{}; //!< For testing
@@ -128,3 +140,62 @@ struct timeval MillisToTimeval(std::chrono::milliseconds ms)
 {
     return MillisToTimeval(count_milliseconds(ms));
 }
+
+std::chrono::nanoseconds ThreadCpuTime()
+{
+#ifdef HAVE_CLOCK_GETTIME
+    // An alternative to clock_gettime() is getrusage().
+
+    timespec t;
+    if (clock_gettime(CLOCK_THREAD_CPUTIME_ID, &t) == -1) {
+        return std::chrono::nanoseconds{0};
+    }
+    return std::chrono::seconds{t.tv_sec} + std::chrono::nanoseconds{t.tv_nsec};
+#elif defined(HAVE_GETTHREADTIMES)
+    // An alternative to GetThreadTimes() is QueryThreadCycleTime() but it
+    // counts CPU cycles.
+
+    FILETIME creation;
+    FILETIME exit;
+    FILETIME kernel;
+    FILETIME user;
+    // GetThreadTimes():
+    // https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-getthreadtimes
+    if (GetThreadTimes(GetCurrentThread(), &creation, &exit, &kernel, &user) == 0) {
+        return std::chrono::nanoseconds{0};
+    }
+
+    // https://learn.microsoft.com/en-us/windows/win32/api/minwinbase/ns-minwinbase-filetime
+    // "... you should copy the low- and high-order parts of the file time to a
+    // ULARGE_INTEGER structure, perform 64-bit arithmetic on the QuadPart
+    // member ..."
+
+    ULARGE_INTEGER kernel_;
+    kernel_.LowPart = kernel.dwLowDateTime;
+    kernel_.HighPart = kernel.dwHighDateTime;
+
+    ULARGE_INTEGER user_;
+    user_.LowPart = user.dwLowDateTime;
+    user_.HighPart = user.dwHighDateTime;
+
+    // https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-getthreadtimes
+    // "Thread kernel mode and user mode times are amounts of time. For example,
+    // if a thread has spent one second in kernel mode, this function will fill
+    // the FILETIME structure specified by lpKernelTime with a 64-bit value of
+    // ten million. That is the number of 100-nanosecond units in one second."
+    return std::chrono::nanoseconds{(kernel_.QuadPart + user_.QuadPart) * 100};
+#else
+    return std::chrono::nanoseconds{0};
+#endif
+}
+
+std::chrono::nanoseconds operator+=(std::atomic<std::chrono::nanoseconds>& a, std::chrono::nanoseconds b)
+{
+    std::chrono::nanoseconds expected;
+    std::chrono::nanoseconds desired;
+    do {
+        expected = a.load();
+        desired = expected + b;
+    } while (!a.compare_exchange_weak(expected, desired));
+    return desired;
+}

src/util/time.h

@@ -6,8 +6,10 @@
 #ifndef BITCOIN_UTIL_TIME_H
 #define BITCOIN_UTIL_TIME_H
 
+#include <atomic>
 #include <chrono> // IWYU pragma: export
 #include <cstdint>
+#include <functional>
 #include <optional>
 #include <string>
 #include <string_view>
@@ -144,4 +146,46 @@ struct timeval MillisToTimeval(int64_t nTimeout);
  */
 struct timeval MillisToTimeval(std::chrono::milliseconds ms);
 
+/**
+ * Retrieve the CPU time (user + system) spent by the current thread.
+ */
+std::chrono::nanoseconds ThreadCpuTime();
+
+/**
+ * Measure CPU time spent by the current thread.
+ * A clock is started when a CpuTimer is created. When the object is destroyed
+ * the elapsed CPU time is calculated and a callback function is invoked,
+ * providing it the elapsed CPU time.
+ */
+class CpuTimer
+{
+public:
+    using FinishedCB = std::function<void(std::chrono::nanoseconds)>;
+
+    /**
+     * Construct a timer.
+     * @param[in] finished_cb A callback to invoke when this object is destroyed.
+     */
+    CpuTimer(const FinishedCB& finished_cb)
+        : m_start{ThreadCpuTime()},
+          m_finished_cb{finished_cb}
+    {
+    }
+
+    ~CpuTimer()
+    {
+        m_finished_cb(ThreadCpuTime() - m_start);
+    }
+
+private:
+    const std::chrono::nanoseconds m_start;
+    FinishedCB m_finished_cb;
+};
+
+/**
+ * Add `b` nanoseconds to a nanoseconds atomic.
+ * @return The value of `a` immediately after the operation.
+ */
+std::chrono::nanoseconds operator+=(std::atomic<std::chrono::nanoseconds>& a, std::chrono::nanoseconds b);
+
 #endif // BITCOIN_UTIL_TIME_H

test/functional/feature_init.py

@@ -88,7 +88,7 @@ class InitTest(BitcoinTestFramework):
 
         args = ['-txindex=1', '-blockfilterindex=1', '-coinstatsindex=1']
         for terminate_line in lines_to_terminate_after:
-            self.log.info(f"Starting node and will exit after line {terminate_line}")
+            self.log.info(f"Starting node and will terminate after line {terminate_line}")
             with node.busy_wait_for_debug_log([terminate_line]):
                 if platform.system() == 'Windows':
                     # CREATE_NEW_PROCESS_GROUP is required in order to be able
@@ -108,12 +108,22 @@ class InitTest(BitcoinTestFramework):
             'blocks/index/*.ldb': 'Error opening block database.',
             'chainstate/*.ldb': 'Error opening coins database.',
             'blocks/blk*.dat': 'Error loading block database.',
+            'indexes/txindex/MANIFEST*': 'LevelDB error: Corruption: CURRENT points to a non-existent file',
+            # Removing these files does not result in a startup error:
+            # 'indexes/blockfilter/basic/*.dat', 'indexes/blockfilter/basic/db/*.*', 'indexes/coinstats/db/*.*',
+            # 'indexes/txindex/*.log', 'indexes/txindex/CURRENT', 'indexes/txindex/LOCK'
         }
 
         files_to_perturb = {
             'blocks/index/*.ldb': 'Error loading block database.',
             'chainstate/*.ldb': 'Error opening coins database.',
             'blocks/blk*.dat': 'Corrupted block database detected.',
+            'indexes/blockfilter/basic/db/*.*': 'LevelDB error: Corruption',
+            'indexes/coinstats/db/*.*': 'LevelDB error: Corruption',
+            'indexes/txindex/*.log': 'LevelDB error: Corruption',
+            'indexes/txindex/CURRENT': 'LevelDB error: Corruption',
+            # Perturbing these files does not result in a startup error:
+            # 'indexes/blockfilter/basic/*.dat', 'indexes/txindex/MANIFEST*', 'indexes/txindex/LOCK'
         }
 
         for file_patt, err_fragment in files_to_delete.items():
@@ -135,9 +145,10 @@ class InitTest(BitcoinTestFramework):
             self.stop_node(0)
 
         self.log.info("Test startup errors after perturbing certain essential files")
+        dirs = ["blocks", "chainstate", "indexes"]
         for file_patt, err_fragment in files_to_perturb.items():
-            shutil.copytree(node.chain_path / "blocks", node.chain_path / "blocks_bak")
-            shutil.copytree(node.chain_path / "chainstate", node.chain_path / "chainstate_bak")
+            for dir in dirs:
+                shutil.copytree(node.chain_path / dir, node.chain_path / f"{dir}_bak")
             target_files = list(node.chain_path.glob(file_patt))
 
             for target_file in target_files:
@@ -151,10 +162,9 @@ class InitTest(BitcoinTestFramework):
 
             start_expecting_error(err_fragment)
 
-            shutil.rmtree(node.chain_path / "blocks")
-            shutil.rmtree(node.chain_path / "chainstate")
-            shutil.move(node.chain_path / "blocks_bak", node.chain_path / "blocks")
-            shutil.move(node.chain_path / "chainstate_bak", node.chain_path / "chainstate")
+            for dir in dirs:
+                shutil.rmtree(node.chain_path / dir)
+                shutil.move(node.chain_path / f"{dir}_bak", node.chain_path / dir)
 
     def init_pid_test(self):
         BITCOIN_PID_FILENAME_CUSTOM = "my_fancy_bitcoin_pid_file.foobar"

test/functional/mempool_accept.py

@@ -45,6 +45,7 @@ from test_framework.util import (
     assert_equal,
     assert_greater_than,
     assert_raises_rpc_error,
+    sync_txindex,
 )
 from test_framework.wallet import MiniWallet
 from test_framework.wallet_util import generate_keypair
@@ -270,6 +271,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
 
         self.log.info('A coinbase transaction')
         # Pick the input of the first tx we created, so it has to be a coinbase tx
+        sync_txindex(self, node)
         raw_tx_coinbase_spent = node.getrawtransaction(txid=node.decoderawtransaction(hexstring=raw_tx_in_block)['vin'][0]['txid'])
         tx = tx_from_hex(raw_tx_coinbase_spent)
         self.check_mempool_result(

test/functional/rpc_net.py

@@ -142,6 +142,7 @@ class NetTest(BitcoinTestFramework):
         # The next two fields will vary for v2 connections because we send a rng-based number of decoy messages
         peer_info.pop("bytesrecv")
         peer_info.pop("bytessent")
+        peer_info.pop("cpu_load", None)
         assert_equal(
             peer_info,
             {

test/functional/rpc_rawtransaction.py

@@ -34,6 +34,7 @@ from test_framework.util import (
     assert_equal,
     assert_greater_than,
     assert_raises_rpc_error,
+    sync_txindex,
 )
 from test_framework.wallet import (
     getnewdestination,
@@ -70,7 +71,7 @@ class RawTransactionsTest(BitcoinTestFramework):
         self.num_nodes = 3
         self.extra_args = [
             ["-txindex"],
-            ["-txindex"],
+            [],
             ["-fastprune", "-prune=1"],
         ]
         # whitelist peers to speed up tx relay / mempool sync
@@ -109,6 +110,7 @@ class RawTransactionsTest(BitcoinTestFramework):
             self.log.info(f"Test getrawtransaction {'with' if n == 0 else 'without'} -txindex")
 
             if n == 0:
+                sync_txindex(self, self.nodes[n])
                 # With -txindex.
                 # 1. valid parameters - only supply txid
                 assert_equal(self.nodes[n].getrawtransaction(txId), tx['hex'])

test/functional/rpc_txoutproof.py

@@ -12,6 +12,7 @@ from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
     assert_raises_rpc_error,
+    sync_txindex,
 )
 from test_framework.wallet import MiniWallet
 
@@ -77,6 +78,7 @@ class MerkleBlockTest(BitcoinTestFramework):
         assert_equal(sorted(self.nodes[0].verifytxoutproof(self.nodes[0].gettxoutproof([txid1, txid2]))), sorted(txlist))
         assert_equal(sorted(self.nodes[0].verifytxoutproof(self.nodes[0].gettxoutproof([txid2, txid1]))), sorted(txlist))
         # We can always get a proof if we have a -txindex
+        sync_txindex(self, self.nodes[1])
         assert_equal(self.nodes[0].verifytxoutproof(self.nodes[1].gettxoutproof([txid_spent])), [txid_spent])
         # We can't get a proof if we specify transactions from different blocks
         assert_raises_rpc_error(-5, "Not all transactions found in specified or retrieved block", self.nodes[0].gettxoutproof, [txid1, txid3])

test/functional/test_framework/util.py

@@ -592,3 +592,10 @@ def find_vout_for_address(node, txid, addr):
         if addr == tx["vout"][i]["scriptPubKey"]["address"]:
             return i
     raise RuntimeError("Vout not found for address: txid=%s, addr=%s" % (txid, addr))
+
+
+def sync_txindex(test_framework, node):
+    test_framework.log.debug("Waiting for node txindex to sync")
+    sync_start = int(time.time())
+    test_framework.wait_until(lambda: node.getindexinfo("txindex")["txindex"]["synced"])
+    test_framework.log.debug(f"Synced in {time.time() - sync_start} seconds")

test/functional/wallet_avoid_mixing_output_types.py

@@ -117,7 +117,6 @@ class AddressInputTypeGrouping(BitcoinTestFramework):
         self.extra_args = [
             [
                 "-addresstype=bech32",
-                "-txindex",
             ],
             [
                 "-addresstype=p2sh-segwit",