wallet: bdb batch 'ErasePrefix', do not create txn internally

Transactions are intended to be started on upper layers rather than
internally by the bdb batch object. This enables us to consolidate
different write operations within a procedure in the same db txn,
improving consistency due to the atomic property of the transaction,
as well as its performance due to the reduction of disk write
operations.

Important Note:
This approach also ensures that the BerkeleyBatch::ErasePrefix
function behaves exactly as the SQLiteBatch::ErasePrefix function,
which does not create a db txn internally.

Furthermore, since the `BerkeleyBatch::ErasePrefix' implementation
erases records one by one (by traversing the db), this change
ensures that the function is always called within an active txn
context. Without this measure, there's a potential risk to consistency;
certain records may be removed while others could persist due to an
internal failure during the procedure.

src/wallet/bdb.cpp

@@ -887,7 +887,12 @@ bool BerkeleyBatch::HasKey(DataStream&& key)
 
 bool BerkeleyBatch::ErasePrefix(Span<const std::byte> prefix)
 {
-    if (!TxnBegin()) return false;
+    // Because this function erases records one by one, ensure that it is executed within a txn context.
+    // Otherwise, consistency is at risk; it's possible that certain records are removed while others
+    // remain due to an internal failure during the procedure.
+    // Additionally, the Dbc::del() cursor delete call below would fail without an active transaction.
+    if (!Assume(activeTxn)) return false;
+
     auto cursor{std::make_unique<BerkeleyCursor>(m_database, *this)};
     // const_cast is safe below even though prefix_key is an in/out parameter,
     // because we are not using the DB_DBT_USERMEM flag, so BDB will allocate
@@ -901,7 +906,7 @@ bool BerkeleyBatch::ErasePrefix(Span<const std::byte> prefix)
         ret = cursor->dbc()->del(0);
     }
     cursor.reset();
-    return TxnCommit() && (ret == 0 || ret == DB_NOTFOUND);
+    return ret == 0 || ret == DB_NOTFOUND;
 }
 
 void BerkeleyDatabase::AddRef()

src/wallet/test/db_tests.cpp

@@ -228,6 +228,39 @@ BOOST_AUTO_TEST_CASE(db_availability_after_write_error)
     }
 }
 
+// Verify 'ErasePrefix' functionality using db keys similar to the ones used by the wallet.
+// Keys are in the form of std::pair<TYPE, ENTRY_ID>
+BOOST_AUTO_TEST_CASE(erase_prefix)
+{
+    const std::string key = "key";
+    const std::string key2 = "key2";
+    const std::string value = "value";
+    const std::string value2 = "value_2";
+    auto make_key = [](std::string type, std::string id) { return std::make_pair(type, id); };
+
+    for (const auto& database : TestDatabases(m_path_root)) {
+        std::unique_ptr<DatabaseBatch> batch = database->MakeBatch();
+
+        // Write two entries with the same key type prefix, a third one with a different prefix
+        // and a fourth one with the type-id values inverted
+        BOOST_CHECK(batch->Write(make_key(key, value), value));
+        BOOST_CHECK(batch->Write(make_key(key, value2), value2));
+        BOOST_CHECK(batch->Write(make_key(key2, value), value));
+        BOOST_CHECK(batch->Write(make_key(value, key), value));
+
+        // Erase the ones with the same prefix and verify result
+        BOOST_CHECK(batch->TxnBegin());
+        BOOST_CHECK(batch->ErasePrefix(DataStream() << key));
+        BOOST_CHECK(batch->TxnCommit());
+
+        BOOST_CHECK(!batch->Exists(make_key(key, value)));
+        BOOST_CHECK(!batch->Exists(make_key(key, value2)));
+        // Also verify that entries with a different prefix were not erased
+        BOOST_CHECK(batch->Exists(make_key(key2, value)));
+        BOOST_CHECK(batch->Exists(make_key(value, key)));
+    }
+}
+
 #ifdef USE_SQLITE
 
 // Test-only statement execution error

src/wallet/test/wallet_tests.cpp

@@ -445,9 +445,11 @@ BOOST_FIXTURE_TEST_CASE(LoadReceiveRequests, TestingSetup)
             auto requests = wallet->GetAddressReceiveRequests();
             auto erequests = {"val_rr11", "val_rr20"};
             BOOST_CHECK_EQUAL_COLLECTIONS(requests.begin(), requests.end(), std::begin(erequests), std::end(erequests));
-            WalletBatch batch{wallet->GetDatabase()};
-            BOOST_CHECK(batch.WriteAddressPreviouslySpent(PKHash(), false));
-            BOOST_CHECK(batch.EraseAddressData(ScriptHash()));
+            RunWithinTxn(wallet->GetDatabase(), /*process_desc*/"test", [](WalletBatch& batch){
+                BOOST_CHECK(batch.WriteAddressPreviouslySpent(PKHash(), false));
+                BOOST_CHECK(batch.EraseAddressData(ScriptHash()));
+                return true;
+            });
         });
         TestLoadWallet(name, format, [](std::shared_ptr<CWallet> wallet) EXCLUSIVE_LOCKS_REQUIRED(wallet->cs_wallet) {
             BOOST_CHECK(!wallet->IsAddressPreviouslySpent(PKHash()));