fix(coinjoin): persist the locks of pending observations added while the record was unreadable

AddPendingObservation() locks its inputs in memory only when the pending-observation record cannot be read or no database transaction can be started. Once the record is readable again nothing ever writes those locks: CheckPendingObservations() writes only the record and only after a release, and the next AddPendingObservation() writes locks only for its own inputs. The record then lists inputs without a persistent lock. After a restart the missing lock reads as a manual unlock, the entry is dropped and the input becomes selectable again while the finalized mixing transaction spending it may still be in flight.

Move the transactional write into PersistPendingObservations(), which writes the record together with the lock of every entry whose coin is still locked, and use it on both the add and the release path. CheckPendingObservations() also runs it when the record was recovered on this pass while entries were already held in memory. Rewriting a lock which is already persisted is idempotent and the set only holds the inputs of recent sessions.

The existing unreadable-record test now re-adds an input before the record is made readable and checks that its lock is persisted by the recovering check pass.

Author: knst

src/coinjoin/client.cpp

@@ -678,6 +678,33 @@ void CCoinJoinClientManager::UpdatedSuccessBlock()
     nCachedLastSuccessBlock = nCachedBlockHeight;
 }
 
+// NOTE: the record which owns the locks and the persistent locks themselves are
+// separate writes and each write is its own implicit transaction, so commit them in
+// one explicit database transaction: a crash in between must neither leave
+// persistently locked coins behind with nothing tracking them (nothing would ever
+// release them again) nor persist the record without its locks (on restart the
+// missing lock reads as a manual unlock, the entry is dropped and the input becomes
+// selectable again while the finalized mixing transaction may still be in flight).
+// If no transaction can be started, don't persist anything rather than risk exactly
+// those partial states. The locks of all entries are written, not only of the ones
+// just added: entries added while nothing could be persisted are locked in memory only.
+// A lock released manually in the meantime is left alone, the next check drops its entry.
+static bool PersistPendingObservations(const CWallet& wallet, wallet::WalletBatch& batch,
+                                       const std::map<COutPoint, int64_t>& pending)
+    EXCLUSIVE_LOCKS_REQUIRED(wallet.cs_wallet)
+{
+    if (!batch.TxnBegin()) return false;
+    bool fPersisted{batch.WriteCoinJoinPendingObs(pending)};
+    for (auto it = pending.begin(); fPersisted && it != pending.end(); ++it) {
+        if (wallet.IsLockedCoin(it->first)) fPersisted = batch.WriteLockedUTXO(it->first);
+    }
+    if (!fPersisted) {
+        batch.TxnAbort();
+        return false;
+    }
+    return batch.TxnCommit();
+}
+
 void CCoinJoinClientManager::AddPendingObservation(const std::vector<COutPoint>& outpoints)
 {
     AssertLockNotHeld(cs_pending_obs);
@@ -698,54 +725,19 @@ void CCoinJoinClientManager::AddPendingObservation(const std::vector<COutPoint>&
         WalletCJLogPrint(m_wallet, "CCoinJoinClientManager::%s -- %s is locked until the finalized mixing transaction is observed\n",
                          __func__, outpoint.ToStringShort());
     }
-    if (!PersistPendingObservations(batch)) {
+    // Never overwrite a record which could not be read (LoadPendingObservations() left
+    // m_pending_obs_loaded unset), that would permanently orphan the locks it still tracks
+    if (!m_pending_obs_loaded || !PersistPendingObservations(*m_wallet, batch, m_pending_obs)) {
         // The in-memory lock still protects these inputs for as long as this process
-        // runs, but a restart before CheckPendingObservations() manages to persist them
-        // would make them selectable again while a valid mixing transaction spending them
-        // may already be in flight. Nothing more we can do about it here beyond making
-        // the failure loud - the wallet database is broken.
+        // runs, but a restart would make them selectable again while a valid mixing
+        // transaction spending them may already be in flight. Nothing we can do about
+        // it here beyond making the failure loud - the wallet database is broken.
         LogPrintf("CCoinJoinClientManager::%s -- ERROR: failed to persist locks for %d successfully mixed input(s), " /* Continued */
-                  "they will not survive a restart until this succeeds\n",
+                  "they will not survive a restart\n",
                   __func__, outpoints.size());
     }
 }
 
-bool CCoinJoinClientManager::PersistPendingObservations(wallet::WalletBatch& batch)
-{
-    AssertLockHeld(m_wallet->cs_wallet);
-    AssertLockHeld(cs_pending_obs);
-
-    // NOTE: the record which owns the locks and the persistent locks themselves are
-    // separate writes and each write is its own implicit transaction, so commit them in
-    // one explicit database transaction: a crash in between must neither leave
-    // persistently locked coins behind with nothing tracking them (nothing would ever
-    // release them again) nor persist the record without its locks (on restart the
-    // missing lock reads as a manual unlock, the entry is dropped and the input becomes
-    // selectable again while the finalized mixing transaction may still be in flight).
-    // If no transaction can be started, don't persist anything rather than risk exactly
-    // those partial states. Likewise if the existing record could not be read
-    // (LoadPendingObservations() left m_pending_obs_loaded unset): overwriting it would
-    // permanently orphan the locks it still tracks.
-    if (!m_pending_obs_loaded || !batch.TxnBegin()) {
-        m_pending_obs_dirty = true;
-        return false;
-    }
-    // Write the locks of all entries, not only of the ones just added: entries added
-    // while nothing could be persisted are locked in memory only. A lock released
-    // manually in the meantime is left alone, the next check drops its entry.
-    bool fPersisted{batch.WriteCoinJoinPendingObs(m_pending_obs)};
-    for (auto it = m_pending_obs.begin(); fPersisted && it != m_pending_obs.end(); ++it) {
-        if (m_wallet->IsLockedCoin(it->first)) fPersisted = m_wallet->LockCoin(it->first, &batch);
-    }
-    if (fPersisted) {
-        fPersisted = batch.TxnCommit();
-    } else {
-        batch.TxnAbort();
-    }
-    m_pending_obs_dirty = !fPersisted;
-    return fPersisted;
-}
-
 void CCoinJoinClientManager::LoadPendingObservations(wallet::WalletBatch& batch)
 {
     AssertLockHeld(cs_pending_obs);
@@ -798,6 +790,9 @@ void CCoinJoinClientManager::CheckPendingObservations(const CTxMemPool& mempool)
     LOCK(m_wallet->cs_wallet);
     LOCK(cs_pending_obs);
 
+    // Entries added while the record could not be read are locked in memory only (see
+    // AddPendingObservation()), persist them once the record has been recovered
+    bool fChanged{!m_pending_obs_loaded && !m_pending_obs.empty()};
     if (!m_pending_obs_loaded) {
         // Read-only, no need to checkpoint the database on the way out
         wallet::WalletBatch batch_load(m_wallet->GetDatabase(), /*_fFlushOnClose=*/false);
@@ -815,7 +810,6 @@ void CCoinJoinClientManager::CheckPendingObservations(const CTxMemPool& mempool)
 
     const int64_t nNow{GetTime()};
     const bool fSynced{m_mn_sync.IsBlockchainSynced()};
-    bool fChanged{false};
     for (auto it = m_pending_obs.begin(); it != m_pending_obs.end();) {
         const COutPoint& outpoint = it->first;
         if (!m_wallet->IsLockedCoin(outpoint)) {
@@ -877,16 +871,7 @@ void CCoinJoinClientManager::CheckPendingObservations(const CTxMemPool& mempool)
     // it may still track locks nothing else would ever release. An entry released
     // above but left in such a record self-heals: once the record is readable again
     // the entry reloads, its coin is no longer locked and it is dropped right here.
-    if (!m_pending_obs_loaded) return;
-    bool fPersisted{true};
-    if (m_pending_obs_dirty) {
-        // Some entries could not be persisted when they were added (e.g. the record could
-        // not be read back then) and are only locked in memory, persist their locks too
-        fPersisted = PersistPendingObservations(get_batch());
-    } else if (fChanged) {
-        fPersisted = get_batch().WriteCoinJoinPendingObs(m_pending_obs);
-    }
-    if (!fPersisted) {
+    if (fChanged && m_pending_obs_loaded && !PersistPendingObservations(*m_wallet, get_batch(), m_pending_obs)) {
         LogPrintf("CCoinJoinClientManager::%s -- ERROR: failed to persist %d pending observation(s)\n", __func__,
                   m_pending_obs.size());
     }

src/coinjoin/client.h

@@ -217,14 +217,9 @@ class CCoinJoinClientManager : public interfaces::CoinJoin::Client
     //! Whether the failure to read the record has been reported already, the read is
     //! retried for as long as this node runs and a corrupt record never becomes readable
     bool m_pending_obs_load_failed GUARDED_BY(cs_pending_obs){false};
-    //! Whether m_pending_obs holds entries whose locks are not persisted yet
-    bool m_pending_obs_dirty GUARDED_BY(cs_pending_obs){false};
 
     /// Populate m_pending_obs from the wallet database, once per run
     void LoadPendingObservations(wallet::WalletBatch& batch) EXCLUSIVE_LOCKS_REQUIRED(cs_pending_obs);
-    /// Persist m_pending_obs along with the locks of its entries in one database transaction
-    bool PersistPendingObservations(wallet::WalletBatch& batch)
-        EXCLUSIVE_LOCKS_REQUIRED(m_wallet->cs_wallet, cs_pending_obs);
 
     // Keep track of current block height
     int nCachedBlockHeight{0};

src/wallet/test/coinjoin_tests.cpp

@@ -22,7 +22,6 @@
 #include <wallet/context.h>
 #include <wallet/db.h>
 #include <wallet/spend.h>
-#include <wallet/test/util.h>
 #include <wallet/wallet.h>
 #include <wallet/walletdb.h>
 
@@ -448,14 +447,18 @@ BOOST_FIXTURE_TEST_CASE(coinjoin_pending_observation_unreadable_tests, CTransact
         }
 
         // Once the record is readable again the entry it tracks is picked back up and the
-        // lock behind it can finally be released
+        // lock behind it can finally be released. An entry added in the meantime is only
+        // locked in memory until then, its lock is persisted along with the record.
+        cj_man.AddPendingObservation({outpointInMemory});
         {
             WalletBatch batch(wallet->GetDatabase());
             BOOST_REQUIRE(batch.WriteCoinJoinPendingObs({{outpointPersisted, nStart}}));
         }
         cj_man.CheckPendingObservations(*m_node.mempool);
         BOOST_CHECK(cj_man.IsPendingObservation(outpointPersisted));
-        BOOST_CHECK_EQUAL(cj_man.GetPendingObservationCount(), 1);
+        BOOST_CHECK_EQUAL(cj_man.GetPendingObservationCount(), 2);
+        BOOST_CHECK(wallet->GetDatabase().MakeBatch()->Exists(
+            std::make_pair(DBKeys::LOCKED_UTXO, std::make_pair(outpointInMemory.hash, outpointInMemory.n))));
 
         m_node.mn_sync->SwitchToNextAsset();
         BOOST_REQUIRE(m_node.mn_sync->IsBlockchainSynced());
@@ -467,63 +470,6 @@ BOOST_FIXTURE_TEST_CASE(coinjoin_pending_observation_unreadable_tests, CTransact
     SetMockTime(0);
 }
 
-BOOST_FIXTURE_TEST_CASE(coinjoin_pending_observation_recovered_tests, CTransactionBuilderTestSetup)
-{
-    // 0.100001 DASH, a valid CoinJoin denomination
-    constexpr CAmount nDenomAmount{10000100};
-    CompactTallyItem tallyItem = GetTallyItem({nDenomAmount, nDenomAmount});
-    const COutPoint outpointPersisted = tallyItem.outpoints[0];
-    const COutPoint outpointInMemory = tallyItem.outpoints[1];
-    const auto has_persistent_lock = [&](WalletDatabase& database, const COutPoint& outpoint) {
-        return database.MakeBatch()->Exists(std::make_pair(DBKeys::LOCKED_UTXO, std::make_pair(outpoint.hash, outpoint.n)));
-    };
-
-    const int64_t nStart{GetTime()};
-    SetMockTime(nStart);
-    BOOST_CHECK(m_node.cj_walletman->doForClient("", [&](CCoinJoinClientManager& cj_man) {
-        cj_man.AddPendingObservation({outpointPersisted});
-    }));
-    BOOST_REQUIRE(wallet->GetDatabase().MakeBatch()->Write(std::string(DBKeys::COINJOIN_PENDING_OBS),
-                                                           std::string("not a pending observation map")));
-    m_node.cj_walletman->removeWallet(wallet->GetName());
-    m_node.cj_walletman->addWallet(wallet);
-
-    BOOST_CHECK(m_node.cj_walletman->doForClient("", [&](CCoinJoinClientManager& cj_man) {
-        // While the record is unreadable a new observation is only locked in memory
-        cj_man.CheckPendingObservations(*m_node.mempool);
-        cj_man.AddPendingObservation({outpointInMemory});
-        BOOST_CHECK(cj_man.IsPendingObservation(outpointInMemory));
-        BOOST_CHECK(WITH_LOCK(wallet->cs_wallet, return wallet->IsLockedCoin(outpointInMemory)));
-        BOOST_CHECK(!has_persistent_lock(wallet->GetDatabase(), outpointInMemory));
-
-        // Once the record is readable again the entries which could not be persisted
-        // before have to be persisted along with their locks
-        {
-            WalletBatch batch(wallet->GetDatabase());
-            BOOST_REQUIRE(batch.WriteCoinJoinPendingObs({{outpointPersisted, nStart}}));
-        }
-        cj_man.CheckPendingObservations(*m_node.mempool);
-        BOOST_CHECK(cj_man.IsPendingObservation(outpointPersisted));
-        BOOST_CHECK(cj_man.IsPendingObservation(outpointInMemory));
-    }));
-
-    // Load a copy of the wallet database, the way a restart would: both inputs must
-    // still be locked and tracked
-    DatabaseOptions options;
-    const auto reloaded = std::make_shared<CWallet>(m_node.chain.get(), /*coinjoin_loader=*/nullptr, "", m_args,
-                                                    DuplicateMockDatabase(wallet->GetDatabase(), options));
-    BOOST_REQUIRE_EQUAL(reloaded->LoadWallet(), DBErrors::LOAD_OK);
-    for (const auto& outpoint : {outpointPersisted, outpointInMemory}) {
-        BOOST_CHECK(has_persistent_lock(reloaded->GetDatabase(), outpoint));
-        BOOST_CHECK(WITH_LOCK(reloaded->cs_wallet, return reloaded->IsLockedCoin(outpoint)));
-    }
-    std::map<COutPoint, int64_t> persisted;
-    BOOST_REQUIRE(WalletBatch(reloaded->GetDatabase()).ReadCoinJoinPendingObs(persisted));
-    BOOST_CHECK_EQUAL(persisted.size(), 2);
-    BOOST_CHECK(persisted.count(outpointInMemory) > 0);
-    SetMockTime(0);
-}
-
 BOOST_FIXTURE_TEST_CASE(coinjoin_manager_start_stop_tests, CTransactionBuilderTestSetup)
 {
     BOOST_CHECK(m_node.cj_walletman->doForClient("", [](auto& cj_man) {