Add lock table and improve kickoff gen

core/src/database/common.rs

@@ -128,19 +128,21 @@ impl Database {
         }
     }
 
-    /// Database function for debugging purposes.
-    pub async fn get_nonce_table(
+    pub async fn lock_operators_kickoff_utxo_table(
         &self,
-        table_name: &str,
-    ) -> Result<Vec<(i32, String, String, String, String)>, BridgeError> {
-        let qr: Vec<(i32, String, String, String, String)> =
-            sqlx::query_as(&format!("SELECT * FROM {};", table_name))
-                .fetch_all(&self.connection)
-                .await?;
-
-        let res: Vec<(i32, String, String, String, String)> = qr.into_iter().collect();
+        // tx: &mut sqlx::Transaction,
+    ) -> Result<(), BridgeError> {
+        sqlx::query("LOCK TABLE operators_kickoff_utxo IN ACCESS EXCLUSIVE MODE;")
+            .execute(&self.connection)
+            .await?;
+        Ok(())
+    }
 
-        Ok(res)
+    pub async fn unlock_operators_kickoff_utxo_table(&self) -> Result<(), BridgeError> {
+        sqlx::query("UNLOCK TABLE operators_kickoff_utxo;")
+            .execute(&self.connection)
+            .await?;
+        Ok(())
     }
 
     /// Operator: If operator already created a kickoff UTXO for this deposit UTXO, return it.
@@ -170,53 +172,47 @@ impl Database {
         &self,
     ) -> Result<Option<UTXO>, BridgeError> {
         // Attempt to fetch the latest transaction details
-        let qr: Option<(TxidDB, i32, i32)> = sqlx::query_as(
-            "SELECT txid, num_kickoffs, cur_unused_kickoff_index FROM deposit_kickoff_generator_txs ORDER BY id DESC LIMIT 1;"
+        let qr: Result<(TxidDB, String, i32), sqlx::Error> = sqlx::query_as(
+            "UPDATE deposit_kickoff_generator_txs
+                SET cur_unused_kickoff_index = cur_unused_kickoff_index + 1
+                WHERE id = (
+                    SELECT id 
+                    FROM deposit_kickoff_generator_txs 
+                    ORDER BY id DESC 
+                    LIMIT 1
+                )
+                RETURNING txid, raw_signed_tx, cur_unused_kickoff_index;",
         )
-        .fetch_optional(&self.connection)
-        .await?;
-
-        // If the table is empty, return None
-        let (txid, num_kickoffs, cur_unused_kickoff_index) = match qr {
-            Some(qr) => qr,
-            None => return Ok(None),
-        };
-
-        // Proceed with the rest of the logic only if there are unused kickoffs
-        if num_kickoffs <= cur_unused_kickoff_index {
-            Ok(None)
-        } else {
-            // Fetch the raw signed transaction
-            let db_transaction = self.begin_transaction().await?;
-            let qr_tx: (String,) = sqlx::query_as(
-                "SELECT raw_signed_tx FROM deposit_kickoff_generator_txs ORDER BY id DESC LIMIT 1;",
-            )
-            .fetch_one(&self.connection)
-            .await?;
-
-            // Deserialize the transaction
-            let tx: bitcoin::Transaction =
-                bitcoin::consensus::deserialize(&hex::decode(qr_tx.0).unwrap())?;
-
-            // Create the outpoint and txout
-            let outpoint = OutPoint {
-                txid: tx.compute_txid(),
-                vout: cur_unused_kickoff_index as u32,
-            };
-            let txout = tx.output[cur_unused_kickoff_index as usize].clone();
-
-            // Update the cur_unused_kickoff_index
-            sqlx::query(
-                "UPDATE deposit_kickoff_generator_txs SET cur_unused_kickoff_index = $1 WHERE txid = $2;"
-            )
-            .bind(cur_unused_kickoff_index + 1)
-            .bind(txid)
-            .execute(&self.connection)
-            .await?;
-            db_transaction.commit().await?;
+        .fetch_one(&self.connection)
+        .await;
 
-            // Return the UTXO
-            Ok(Some(UTXO { outpoint, txout }))
+        match qr {
+            Ok((txid, raw_signed_tx, cur_unused_kickoff_index)) => {
+                // Deserialize the transaction
+                let tx: bitcoin::Transaction =
+                    bitcoin::consensus::deserialize(&hex::decode(raw_signed_tx).unwrap())?;
+
+                // Create the outpoint and txout
+                let outpoint = OutPoint {
+                    txid: txid.0,
+                    vout: cur_unused_kickoff_index as u32,
+                };
+                let txout = tx.output[cur_unused_kickoff_index as usize].clone();
+
+                Ok(Some(UTXO { outpoint, txout }))
+            }
+            Err(sqlx::Error::RowNotFound) => Ok(None),
+            Err(e) => {
+                let db_error = e.as_database_error().ok_or(BridgeError::PgDatabaseError(
+                    "THIS SHOULD NOT HAPPEN".to_string(),
+                ))?;
+                // if error is 23514, it means there is no more unused kickoffs
+                if db_error.is_check_violation() {
+                    Ok(None)
+                } else {
+                    Err(BridgeError::PgDatabaseError(db_error.to_string()))
+                }
+            }
         }
     }
 

core/src/errors.rs

@@ -121,6 +121,9 @@ pub enum BridgeError {
     /// Database error
     #[error("DatabaseError: {0}")]
     DatabaseError(#[from] sqlx::Error),
+    /// Database error
+    #[error("PgDatabaseError: {0}")]
+    PgDatabaseError(String),
     /// Operator tries to claim with different bridge funds with the same withdrawal idx
     #[error("AlreadySpentWithdrawal")]
     AlreadySpentWithdrawal,

core/src/operator.rs

@@ -114,6 +114,8 @@ where
 
         let tx = self.db.begin_transaction().await?;
 
+        self.db.lock_operators_kickoff_utxo_table().await?;
+
         // 2. Check if we alredy created a kickoff UTXO for this deposit UTXO
         let kickoff_utxo = self.db.get_kickoff_utxo(deposit_outpoint).await?;
         // if we already have a kickoff UTXO for this deposit UTXO, return it
@@ -134,14 +136,21 @@ where
                 .signer
                 .sign(TapSighash::from_byte_array(kickoff_sig_hash));
 
+            self.db.unlock_operators_kickoff_utxo_table().await?;
             tx.commit().await?;
-
             return Ok((kickoff_utxo, sig));
         }
 
         // Check if we already have an unused kickoff UTXO available
         let unused_kickoff_utxo = self.db.get_unused_kickoff_utxo_and_increase_idx().await?;
         if let Some(unused_kickoff_utxo) = unused_kickoff_utxo {
+            self.db
+                .save_kickoff_utxo(deposit_outpoint, unused_kickoff_utxo.clone())
+                .await?;
+
+            self.db.unlock_operators_kickoff_utxo_table().await?;
+            tx.commit().await?;
+
             tracing::debug!(
                 "Unused kickoff UTXO found: {:?} found for deposit UTXO: {:?}",
                 unused_kickoff_utxo,
@@ -154,15 +163,10 @@ where
                 unused_kickoff_utxo.outpoint.vout.to_be_bytes()
             );
 
-            self.db
-                .save_kickoff_utxo(deposit_outpoint, unused_kickoff_utxo.clone())
-                .await?;
-
             let sig = self
                 .signer
                 .sign(TapSighash::from_byte_array(kickoff_sig_hash));
 
-            tx.commit().await?;
             Ok((unused_kickoff_utxo, sig))
         } else {
             // 3. Create a kickoff transaction but do not broadcast it
@@ -247,17 +251,6 @@ where
                 deposit_outpoint
             );
 
-            let kickoff_sig_hash = crate::sha256_hash!(
-                deposit_outpoint.txid,
-                deposit_outpoint.vout.to_be_bytes(),
-                kickoff_utxo.outpoint.txid,
-                kickoff_utxo.outpoint.vout.to_be_bytes()
-            );
-
-            let sig = self
-                .signer
-                .sign(TapSighash::from_byte_array(kickoff_sig_hash));
-
             // In a db tx, save the kickoff_utxo for this deposit_outpoint
             // and update the db with the new funding_utxo as the change
 
@@ -281,6 +274,17 @@ where
 
             tx.commit().await?;
 
+            let kickoff_sig_hash = crate::sha256_hash!(
+                deposit_outpoint.txid,
+                deposit_outpoint.vout.to_be_bytes(),
+                kickoff_utxo.outpoint.txid,
+                kickoff_utxo.outpoint.vout.to_be_bytes()
+            );
+
+            let sig = self
+                .signer
+                .sign(TapSighash::from_byte_array(kickoff_sig_hash));
+
             Ok((kickoff_utxo, sig))
         }
     }

scripts/schema.sql

@@ -69,10 +69,11 @@ create table if not exists deposit_kickoff_utxos (
 -- Operator table for kickoff utxo and funding utxo for deposits
 /* This table holds the funding utxos sent by the operators for each deposit. */
 create table if not exists deposit_kickoff_generator_txs (
-    txid text primary key not null check (txid ~ '^[a-fA-F0-9]{64}'),
+    id serial primary key,
+    txid text unique not null check (txid ~ '^[a-fA-F0-9]{64}'),
     raw_signed_tx text not null,
     num_kickoffs int not null,
-    cur_unused_kickoff_index int not null,
+    cur_unused_kickoff_index int not null check (cur_unused_kickoff_index <= num_kickoffs),
     funding_txid text not null check (funding_txid ~ '^[a-fA-F0-9]{64}'),
     created_at timestamp not null default now()
 );