Fixed race condition

Bug: https://github.com/cashubtc/cdk/actions/runs/15683152414/job/44190084378?pr=822#step:5:19212

Reason: a race condition between removing proofs while melting and the quote states being updated.

Solution:

1. Error on duplicate proofs
2. Read quote when updating to avoid race conditions and rollbacks

Real solution: A transaction trait in the storage layer. That is coming next

crates/cdk-common/src/database/mint/mod.rs

@@ -88,7 +88,7 @@ pub trait QuotesDatabase {
         &self,
         quote_id: &Uuid,
         state: MeltQuoteState,
-    ) -> Result<MeltQuoteState, Self::Err>;
+    ) -> Result<(MeltQuoteState, mint::MeltQuote), Self::Err>;
     /// Get all [`mint::MeltQuote`]s
     async fn get_melt_quotes(&self) -> Result<Vec<mint::MeltQuote>, Self::Err>;
     /// Remove [`mint::MeltQuote`]

crates/cdk-common/src/database/mod.rs

@@ -22,6 +22,11 @@ pub enum Error {
     /// Database Error
     #[error(transparent)]
     Database(Box<dyn std::error::Error + Send + Sync>),
+
+    /// Duplicate entry
+    #[error("Duplicate entry")]
+    Duplicate,
+
     /// DHKE error
     #[error(transparent)]
     DHKE(#[from] crate::dhke::Error),

crates/cdk-redb/src/mint/mod.rs

@@ -470,12 +470,13 @@ impl MintQuotesDatabase for MintRedbDatabase {
         &self,
         quote_id: &Uuid,
         state: MeltQuoteState,
-    ) -> Result<MeltQuoteState, Self::Err> {
+    ) -> Result<(MeltQuoteState, mint::MeltQuote), Self::Err> {
         let write_txn = self.db.begin_write().map_err(Error::from)?;
 
         let current_state;
+        let mut melt_quote: mint::MeltQuote;
+
         {
-            let mut melt_quote: mint::MeltQuote;
             let mut table = write_txn
                 .open_table(MELT_QUOTES_TABLE)
                 .map_err(Error::from)?;
@@ -506,7 +507,7 @@ impl MintQuotesDatabase for MintRedbDatabase {
         }
         write_txn.commit().map_err(Error::from)?;
 
-        Ok(current_state)
+        Ok((current_state, melt_quote))
     }
 
     async fn get_melt_quotes(&self) -> Result<Vec<mint::MeltQuote>, Self::Err> {

crates/cdk-sqlite/src/mint/async_rusqlite.rs

@@ -4,7 +4,7 @@ use std::sync::{mpsc as std_mpsc, Arc, Mutex};
 use std::thread::spawn;
 use std::time::Instant;
 
-use rusqlite::{Connection, TransactionBehavior};
+use rusqlite::{ffi, Connection, ErrorCode, TransactionBehavior};
 use tokio::sync::{mpsc, oneshot};
 
 use crate::common::SqliteConnectionManager;
@@ -202,6 +202,26 @@ fn rusqlite_spawn_worker_threads(
                     Ok(ok) => reply_to.send(ok),
                     Err(err) => {
                         tracing::error!("Failed query with error {:?}", err);
+                        let err = if let Error::Sqlite(rusqlite::Error::SqliteFailure(
+                            ffi::Error {
+                                code,
+                                extended_code,
+                            },
+                            _,
+                        )) = &err
+                        {
+                            if *code == ErrorCode::ConstraintViolation
+                                && (*extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
+                                    || *extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
+                            {
+                                Error::Duplicate
+                            } else {
+                                err
+                            }
+                        } else {
+                            err
+                        };
+
                         reply_to.send(DbResponse::Error(err))
                     }
                 };
@@ -331,6 +351,27 @@ fn rusqlite_worker_manager(
                                         tx_id,
                                         err
                                     );
+                                    let err = if let Error::Sqlite(
+                                        rusqlite::Error::SqliteFailure(
+                                            ffi::Error {
+                                                code,
+                                                extended_code,
+                                            },
+                                            _,
+                                        ),
+                                    ) = &err
+                                    {
+                                        if *code == ErrorCode::ConstraintViolation
+                                            && (*extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
+                                                || *extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
+                                        {
+                                            Error::Duplicate
+                                        } else {
+                                            err
+                                        }
+                                    } else {
+                                        err
+                                    };
                                     reply_to.send(DbResponse::Error(err))
                                 }
                             };

crates/cdk-sqlite/src/mint/error.rs

@@ -9,6 +9,10 @@ pub enum Error {
     #[error(transparent)]
     Sqlite(#[from] rusqlite::Error),
 
+    /// Duplicate entry
+    #[error("Record already exists")]
+    Duplicate,
+
     /// Pool error
     #[error(transparent)]
     Pool(#[from] crate::pool::Error<rusqlite::Error>),
@@ -98,6 +102,9 @@ pub enum Error {
 
 impl From<Error> for cdk_common::database::Error {
     fn from(e: Error) -> Self {
-        Self::Database(Box::new(e))
+        match e {
+            Error::Duplicate => Self::Duplicate,
+            e => Self::Database(Box::new(e)),
+        }
     }
 }

crates/cdk-sqlite/src/mint/mod.rs

@@ -674,10 +674,10 @@ ON CONFLICT(request_lookup_id) DO UPDATE SET
         &self,
         quote_id: &Uuid,
         state: MeltQuoteState,
-    ) -> Result<MeltQuoteState, Self::Err> {
+    ) -> Result<(MeltQuoteState, mint::MeltQuote), Self::Err> {
         let transaction = self.pool.begin().await?;
 
-        let quote = query(
+        let mut quote = query(
             r#"
             SELECT
                 id,
@@ -732,7 +732,10 @@ ON CONFLICT(request_lookup_id) DO UPDATE SET
             }
         };
 
-        Ok(quote.state)
+        let old_state = quote.state;
+        quote.state = state;
+
+        Ok((old_state, quote))
     }
 
     async fn remove_melt_quote(&self, quote_id: &Uuid) -> Result<(), Self::Err> {
@@ -816,7 +819,7 @@ impl MintProofsDatabase for MintSqliteDatabase {
         for proof in proofs {
             query(
                 r#"
-                INSERT OR IGNORE INTO proof
+                INSERT INTO proof
                 (y, amount, keyset_id, secret, c, witness, state, quote_id, created_time)
                 VALUES
                 (:y, :amount, :keyset_id, :secret, :c, :witness, :state, :quote_id, :created_time)
@@ -852,10 +855,11 @@ impl MintProofsDatabase for MintSqliteDatabase {
 
         let total_deleted = query(
             r#"
-            DELETE FROM proof WHERE y IN (:ys) AND state != 'SPENT'
+            DELETE FROM proof WHERE y IN (:ys) AND state NOT IN (:exclude_state)
             "#,
         )
         .bind_vec(":ys", ys.iter().map(|y| y.to_bytes().to_vec()).collect())
+        .bind_vec(":exclude_state", vec![State::Spent.to_string()])
         .execute(&transaction)
         .await?;
 
@@ -974,7 +978,11 @@ impl MintProofsDatabase for MintSqliteDatabase {
 
         if current_states.len() != ys.len() {
             transaction.rollback().await?;
-            tracing::warn!("Attempted to update state of non-existent proof");
+            tracing::warn!(
+                "Attempted to update state of non-existent proof {} {}",
+                current_states.len(),
+                ys.len()
+            );
             return Err(database::Error::ProofNotFound);
         }
 

crates/cdk/src/mint/melt.rs

@@ -297,7 +297,7 @@ impl Mint {
         &self,
         melt_request: &MeltRequest<Uuid>,
     ) -> Result<MeltQuote, Error> {
-        let state = self
+        let (state, quote) = self
             .localstore
             .update_melt_quote_state(melt_request.quote(), MeltQuoteState::Pending)
             .await?;
@@ -309,12 +309,6 @@ impl Mint {
             MeltQuoteState::Unknown => Err(Error::UnknownPaymentState),
         }?;
 
-        let quote = self
-            .localstore
-            .get_melt_quote(melt_request.quote())
-            .await?
-            .ok_or(Error::UnknownQuote)?;
-
         self.pubsub_manager
             .melt_quote_status(&quote, None, None, MeltQuoteState::Pending);
 
@@ -347,9 +341,19 @@ impl Mint {
             ));
         }
 
-        self.localstore
+        if let Some(err) = self
+            .localstore
             .add_proofs(melt_request.inputs().clone(), None)
-            .await?;
+            .await
+            .err()
+        {
+            match err {
+                cdk_common::database::Error::Duplicate => {
+                    // the proofs already exits, it will be errored by `check_ys_spendable`
+                }
+                err => return Err(Error::Database(err)),
+            }
+        }
 
         self.check_ys_spendable(&input_ys, State::Pending).await?;
         for proof in melt_request.inputs() {

crates/cdk/src/mint/swap.rs

@@ -24,9 +24,19 @@ impl Mint {
 
         // After swap request is fully validated, add the new proofs to DB
         let input_ys = swap_request.inputs().ys()?;
-        self.localstore
+        if let Some(err) = self
+            .localstore
             .add_proofs(swap_request.inputs().clone(), None)
-            .await?;
+            .await
+            .err()
+        {
+            match err {
+                cdk_common::database::Error::Duplicate => {
+                    // the proofs already exits, it will be errored by `check_ys_spendable`
+                }
+                err => return Err(Error::Database(err)),
+            }
+        }
         self.check_ys_spendable(&input_ys, State::Pending).await?;
 
         let mut promises = Vec::with_capacity(swap_request.outputs().len());