Merge bitcoin/bitcoin#34931: validation: abort on DB unreadable coins instead of treating them as missing

### ci/test/00_setup_env_native_previous_releases.sh
@@ -9,8 +9,9 @@ export LC_ALL=C.UTF-8
 export CONTAINER_NAME=ci_native_previous_releases
 export CI_IMAGE_NAME_TAG="mirror.gcr.io/ubuntu:22.04"
 # Use minimum supported python3.10 and gcc-12, see doc/dependencies.md
-export PACKAGES="gcc-12 g++-12 python3-zmq"
+export PACKAGES="gcc-12 g++-12 python3-zmq libleveldb-dev python3-pip"
 export DEP_OPTS="CC=gcc-12 CXX=g++-12"
+export PIP_PACKAGES="plyvel"
 export TEST_RUNNER_EXTRA="--previous-releases --coverage --extended --exclude feature_dbcrash"  # Run extended tests so that coverage does not fail, but exclude the very slow dbcrash
 export GOAL="install"
 export CI_LIMIT_STACK_SIZE=1

### src/dbwrapper.h
@@ -11,6 +11,7 @@
 #include <streams.h>
 #include <util/byte_units.h>
 #include <util/check.h>
+#include <util/expected.h>
 #include <util/fs.h>
 #include <util/obfuscation.h>
 
@@ -216,26 +217,84 @@ class CDBWrapper
     CDBWrapper(const CDBWrapper&) = delete;
     CDBWrapper& operator=(const CDBWrapper&) = delete;
 
+    struct ReadFailure {
+        enum class Code {
+            DeserializationError,   //!< Key exists but value could not be deserialized.
+            DatabaseError,          //!< Unexpected internal DB error.
+        };
+
+        Code status;
+        std::string err_msg;
+    };
+
+    using ReadStatus = util::Expected<bool, ReadFailure>;
+
+    /**
+     * Read and deserialize a value from the database, with explicit error discrimination.
+     *
+     * Unlike Read(), this method distinguishes between a missing key, a deserialization
+     * failure (DeserializationError), and an internal DB error (DatabaseError),
+     * enabling callers to treat data corruption differently from an absent entry.
+     *
+     * @note Callers are expected to provide well-formed keys; key serialization
+     *       is the only operation that may throw.
+     *
+     * @param[in]  key    The key to look up.
+     * @param[out] value  Populated with the deserialized value when the returned
+     *                    Expected holds true; indeterminate otherwise.
+     * @return On success, true if the key was found (value populated) or false if
+     *         the key was absent. On failure, a ReadFailure describing the error.
+     */
     template <typename K, typename V>
-    bool Read(const K& key, V& value) const
+    [[nodiscard]] ReadStatus TryRead(const K& key, V& value) const
     {
         DataStream ssKey{};
         ssKey.reserve(DBWRAPPER_PREALLOC_KEY_SIZE);
+        // Key serialization is the only operation that may throw.
+        // Callers are expected to provide well-formed keys.
         ssKey << key;
-        std::optional<std::string> strValue{ReadImpl(ssKey)};
-        if (!strValue) {
-            return false;
+
+        std::optional<std::string> strValue;
+        try {
+            strValue = ReadImpl(ssKey);
+            if (!strValue) {
+                return false; // not found
+            }
+        } catch (const std::exception& e) {
+            return util::Unexpected(ReadFailure{ReadFailure::Code::DatabaseError, e.what()});
         }
+
         try {
             std::span ssValue{MakeWritableByteSpan(*strValue)};
             m_obfuscation(ssValue);
             SpanReader{ssValue} >> value;
-        } catch (const std::exception&) {
-            return false;
+        } catch (const std::exception& e) {
+            return util::Unexpected(ReadFailure{ReadFailure::Code::DeserializationError, e.what()});
         }
+
         return true;
     }
 
+    /**
+     * Wrapper around TryRead() that preserves the original Read() semantics:
+     * returns true on success, false if the key is absent or deserialization
+     * fails, and throws dbwrapper_error on an internal DB error.
+     *
+     * Prefer TryRead() when the caller needs to distinguish between a missing
+     * key and a corrupt value.
+     */
+    template <typename K, typename V>
+    bool Read(const K& key, V& value) const
+    {
+        const ReadStatus res = TryRead(key,value);
+        if (res.has_value()) return res.value();
+        switch (const auto& [err_code, err_msg] = res.error(); err_code) {
+            case ReadFailure::Code::DeserializationError: return false;
+            case ReadFailure::Code::DatabaseError: throw dbwrapper_error(err_msg);
+        } // no default case, so the compiler can warn about missing cases
+        std::abort(); // unreachable
+    }
+
     template <typename K, typename V>
     void Write(const K& key, const V& value, bool fSync = false)
     {

### src/test/dbwrapper_tests.cpp
@@ -10,6 +10,7 @@
 #include <util/byte_units.h>
 #include <util/string.h>
 
+#include <fstream>
 #include <memory>
 #include <ranges>
 
@@ -194,6 +195,110 @@ BOOST_AUTO_TEST_CASE(dbwrapper_batch)
     }
 }
 
+// Verify that Read() returns false (without throwing) when the stored value
+// fails to be deserialized
+BOOST_AUTO_TEST_CASE(dbwrapper_read_returns_false_on_deserialization_error)
+{
+    for (const bool obfuscate : {false, true}) {
+        const fs::path path{m_args.GetDataDirBase() / (obfuscate ? "dbwrapper_deser_obf" : "dbwrapper_deser_noobf")};
+        CDBWrapper dbw({.path = path, .cache_bytes = 1 << 20, .wipe_data = true, .obfuscate = obfuscate});
+
+        constexpr uint8_t key{'X'};
+
+        // Write a single byte. uint256 requires 32 bytes, so reading this key
+        // as uint256 must trigger a deserialization error inside Read()
+        dbw.Write(key, uint8_t{0xFF});
+        BOOST_CHECK(dbw.Exists(key));
+
+        // Read() must catch the deserialization exception and return false,
+        // the same as if the key were absent
+        uint256 result;
+        BOOST_CHECK(!dbw.Read(key, result));
+    }
+}
+
+// Verify Read() throws dbwrapper_error due to an internal db error
+BOOST_AUTO_TEST_CASE(dbwrapper_read_throws_on_db_error)
+{
+    const fs::path path{m_args.GetDataDirBase() / "dbwrapper_db_error"};
+    constexpr uint8_t key{'Y'};
+
+    const auto make_db = [] (const fs::path& path, const bool force_compact) {
+        return CDBWrapper({.path = path, .cache_bytes = 1 << 20, .obfuscate = false,
+                        .options = {.force_compact = force_compact}});
+    };
+
+    // Write a value and close the database
+    make_db(path, /*force_compact=*/false).Write(key, m_rng.rand256());
+
+    // Force compaction to ensure the data is written into the .ldb files
+    // rather than left in the WAL.
+    (void)make_db(path, /*force_compact=*/true);
+
+    // Corrupt every table so any subsequent Read() fails
+    for (const auto& entry : fs::directory_iterator(path)) {
+        if (entry.path().extension() == ".ldb") {
+            std::ofstream{entry.path(), std::ios::binary | std::ios::trunc}
+            .write("\xff", 1);
+        }
+    }
+
+    // Read() should detect the issue now and throw
+    const auto db{make_db(path, /*force_compact=*/false)};
+    uint256 result;
+    BOOST_CHECK_EXCEPTION(db.Read(key, result), dbwrapper_error, HasReason("Fatal LevelDB error"));
+
+    // TryRead() must return DatabaseError (without throwing).
+    CDBWrapper::ReadStatus status = db.TryRead(key, result);
+    BOOST_REQUIRE(!status);
+    BOOST_CHECK(status.error().status == CDBWrapper::ReadFailure::Code::DatabaseError);
+    BOOST_CHECK(status.error().err_msg.find("Fatal LevelDB error") != std::string::npos);
+}
+
+// Exercise TryRead() return values directly: found, absent and DeserializationError.
+// DatabaseError is tested inside 'dbwrapper_read_throws_on_db_error' test
+BOOST_AUTO_TEST_CASE(dbwrapper_tryread)
+{
+    for (const bool obfuscate : {false, true}) {
+        const fs::path path{m_args.GetDataDirBase() / (obfuscate ? "dbwrapper_tryread_obf" : "dbwrapper_tryread_noobf")};
+        CDBWrapper dbw({.path = path, .cache_bytes = 1 << 20, .wipe_data = true, .obfuscate = obfuscate});
+
+        constexpr uint8_t key_ok{'A'};
+        constexpr uint8_t key_missing{'B'};
+        constexpr uint8_t key_bad{'C'};
+
+        uint256 written_value{m_rng.rand256()};
+        dbw.Write(key_ok, written_value);
+        dbw.Write(key_bad, uint8_t{0xFF});
+
+        // Found: key exists, value deserializes correctly
+        {
+            uint256 read_value;
+            CDBWrapper::ReadStatus status = dbw.TryRead(key_ok, read_value);
+            BOOST_REQUIRE(status);
+            BOOST_CHECK(status.value());
+            BOOST_CHECK_EQUAL(read_value, written_value);
+        }
+
+        // Absent: key does not exist
+        {
+            uint256 read_value;
+            CDBWrapper::ReadStatus status = dbw.TryRead(key_missing, read_value);
+            BOOST_REQUIRE(status);
+            BOOST_CHECK(!status.value());
+        }
+
+        // DeserializationError: key exists but stored value is too short
+        {
+            uint256 read_value;
+            CDBWrapper::ReadStatus status = dbw.TryRead(key_bad, read_value);
+            BOOST_REQUIRE(!status);
+            BOOST_CHECK(status.error().status == CDBWrapper::ReadFailure::Code::DeserializationError);
+            BOOST_CHECK(!status.error().err_msg.empty());
+        }
+    }
+}
+
 BOOST_AUTO_TEST_CASE(dbwrapper_iterator)
 {
     // Perform tests both obfuscated and non-obfuscated.

### src/txdb.cpp
@@ -87,11 +87,24 @@ void CCoinsViewDB::ResizeCache(size_t new_cache_size)
 
 std::optional<Coin> CCoinsViewDB::GetCoin(const COutPoint& outpoint) const
 {
-    if (Coin coin; m_db->Read(CoinEntry(&outpoint), coin)) {
-        Assert(!coin.IsSpent()); // The UTXO database should never contain spent coins
-        return coin;
+    Coin coin;
+    const CDBWrapper::ReadStatus res = m_db->TryRead(CoinEntry(&outpoint), coin);
+    if (!res) {
+        // Propagate errors so CCoinsViewErrorCatcher triggers a clean shutdown.
+        switch (const auto& [err_code, err_msg] = res.error(); err_code) {
+            case CDBWrapper::ReadFailure::Code::DeserializationError:
+                throw dbwrapper_error{strprintf("Coin deserialization failure: %s", err_msg)};
+            case CDBWrapper::ReadFailure::Code::DatabaseError:
+                throw dbwrapper_error{strprintf("Coin DB read failure: %s", err_msg)};
+        } // no default case, so the compiler can warn about missing cases
+        std::abort(); // unreachable
     }
-    return std::nullopt;
+
+    // Check whether the coin exists
+    if (!res.value()) return std::nullopt;
+    // Coin found, ensure UTXO database never contains spent coins
+    Assert(!coin.IsSpent());
+    return coin;
 }
 
 std::optional<Coin> CCoinsViewDB::PeekCoin(const COutPoint& outpoint) const

### test/functional/feature_utxo_abort_on_error.py
@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+# Copyright (c) The Bitcoin Core developers
+# Distributed under the MIT software license.
+
+"""
+Ensures that UTXO unserialization errors abort the node, and does not
+cause a consensus divergence.
+
+A valid UTXO is created and shared between two nodes. The raw database
+entry is then deliberately modified on one node so it can no longer be
+deserialized. When the other node spends that UTXO and mines a block,
+the node with the unserializable entry must abort during block connection
+rather than silently treating the coin as absent and marking the block
+BLOCK_FAILED_VALID, which would cause it to permanently diverge from the
+network's best chain.
+"""
+
+try:
+    import plyvel  # type: ignore[import]
+except ImportError:
+    plyvel = None
+
+from test_framework.blocktools import COINBASE_MATURITY
+from test_framework.test_framework import BitcoinTestFramework, SkipTest
+from test_framework.util import assert_equal
+from test_framework.wallet import MiniWallet
+
+
+class UTXOAbortOnErrorTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.setup_clean_chain = True
+        self.num_nodes = 2
+
+    def skip_test_if_missing_module(self):
+        if plyvel is None:
+            raise SkipTest("plyvel not available (pip install plyvel)")
+
+    def setup_network(self):
+        self.setup_nodes() # Start with nodes disconnected
+
+    def run_test(self):
+        node0, node1 = self.nodes
+
+        self.log.info("Mining mature coinbase on node0")
+        wallet0 = MiniWallet(node0)
+        self.generate(wallet0, COINBASE_MATURITY + 1, sync_fun=self.no_op)
+        assert_equal(node0.getblockcount(), COINBASE_MATURITY + 1)
+
+        # The coinbase of block 1 is now mature. This is the UTXO we will
+        # make unserializable on node0 and spend on node1.
+        coinbase_txid = node0.getblock(node0.getblockhash(1))['tx'][0]
+        assert node0.gettxout(coinbase_txid, 0) is not None, f"Expected UTXO {coinbase_txid}:0 to exist in UTXO set"
+
+        self.log.info("Preparing a spend of the mature coinbase UTXO (not broadcast)")
+        utxo = wallet0.get_utxo(txid=coinbase_txid, vout=0, mark_as_spent=False)
+        spend_tx_hex = wallet0.create_self_transfer(utxo_to_spend=utxo)['hex']
+
+        self.log.info("Sync node1 up to the tip, then isolate nodes")
+        self.connect_nodes(0, 1)
+        self.sync_blocks()
+        assert_equal(node1.getblockcount(), COINBASE_MATURITY + 1)
+        self.disconnect_nodes(0, 1)
+
+        self.log.info("Make UTXO unserializable in node0 database")
+        self.stop_node(0)
+
+        # LevelDB key for the CoinEntry serialization:
+        # key = DB_COIN (0x43='C') || txid (32 bytes) || VARINT(vout=0)
+        coin_key = b'\x43' + bytes.fromhex(coinbase_txid)[::-1] + b'\x00'
+        chainstate_path = str(node0.chain_path / "chainstate")
+
+        # Update entry to mimic an incompatible serialization format
+        with plyvel.DB(chainstate_path, create_if_missing=False, compression=None) as db:
+            existing_value = db.get(coin_key)
+            assert existing_value is not None, f"UTXO {coinbase_txid}:0 not found in db"
+
+            # Write a single-byte value. After XOR deobfuscation this is still just one
+            # byte, far too short for a valid Coin (which needs height VARINT + amount
+            # VARINT + script at minimum). Deserialization will throw "end of data" when
+            # trying to read beyond the first field.
+            db.put(coin_key, b'\x00')
+
+        self.log.info("Restart node0 — the unserializable entry is only visible during block validation, not at startup")
+        self.start_node(0)
+
+        # Individual coin values are not read at startup; only block validation
+        # touches them. The node starts cleanly regardless of the tampered entry.
+        assert_equal(node0.getblockcount(), COINBASE_MATURITY + 1)
+
+        # Now spend the corrupted UTXO
+        self.log.info("node1 broadcasts the spend and mines a block")
+        node1.sendrawtransaction(spend_tx_hex)
+        spending_block_hash = self.generate(node1, 1, sync_fun=self.no_op)[0]
+        assert_equal(node1.getblockcount(), COINBASE_MATURITY + 2)
+
+        self.log.info("Connect node0 to node1 — node0 must abort when it tries to connect the spending block")
+        # Verify that the unserializable entry triggers the expected error and is
+        # never silently misreported as a missing input (bad-txns-inputs-missingorspent),
+        # which would indicate the previous silent-divergence behaviour.
+        with node0.assert_debug_log(expected_msgs=["Error reading from database: Coin deserialization failure"],
+                                    unexpected_msgs=["bad-txns-inputs-missingorspent"]):
+            try:
+                self.connect_nodes(0, 1)
+            except Exception:
+                pass # node0 may validly abort before connect_nodes returns
+
+            # Confirm node0 aborted with SIGABRT
+            self.wait_until(lambda: self.nodes[0].is_node_stopped(
+                expected_ret_code=-6,
+                expected_stderr="Error: Error reading from database, shutting down.",
+            ))
+
+        self.log.info("node0 aborted cleanly — no silent divergence occurred")
+
+        self.log.info("Restart node0 and verify the spending block was not permanently marked invalid")
+        self.start_node(0)
+        assert_equal(node0.getblockcount(), COINBASE_MATURITY + 1)
+
+        # If BLOCK_FAILED_VALID had been written to disk, the node would be
+        # permanently stuck on a stale tip even after the tampered entry is resolved.
+        tips = node0.getchaintips()
+        permanently_invalid = [
+            t for t in tips
+            if t['hash'] == spending_block_hash and t['status'] == 'invalid'
+        ]
+        assert len(permanently_invalid) == 0, f"Spending block {spending_block_hash} must not be marked BLOCK_FAILED_VALID"
+
+
+if __name__ == '__main__':
+    UTXOAbortOnErrorTest(__file__).main()

### test/functional/test_runner.py
@@ -90,6 +90,7 @@
     'feature_pruning.py',
     'feature_dbcrash.py',
     'feature_index_prune.py',
+    'feature_utxo_abort_on_error.py',
 ]
 
 # Special script to run each bench sanity check