improve monitoring with deletion from db

taptrade-cli-demo/coordinator/.cookie

@@ -1 +1 @@
-__cookie__:db7e210a348561c1afa367f9f7f021fdb74e0fc131944fb1fc892239c651184b
+__cookie__:b3219c105fc87f4de97f8a14a17ea82da2f0e5c17ba79bbf45e641de96cd6a55

taptrade-cli-demo/coordinator/src/coordinator/monitoring.rs

@@ -3,9 +3,10 @@
 // prevent querying the db all the time.
 // Also needs to implement punishment logic in case a fraud is detected.
 use super::*;
+use anyhow::Context;
 use sha2::{Digest, Sha256};
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub enum Table {
 	Orderbook,
 	ActiveTrades,
@@ -28,6 +29,14 @@ impl MonitoringBond {
 		Ok(sha256(&hex::decode(&self.bond_tx_hex)?))
 	}
 
+	async fn remove_from_db_tables(&self, db: Arc<CoordinatorDB>) -> Result<()> {
+		// remove bond from db
+		db.remove_violating_bond(self)
+			.await
+			.context("Error removing violating bond from db")?;
+		Ok(())
+	}
+
 	// the current implementation only publishes the bond and removes the offer from the db
 	// in a more advanced implementation we could increase the transaction fee (cpfp) and
 	// continue monitoring the bond transaction until a confirmation happens for maximum pain
@@ -35,11 +44,14 @@ impl MonitoringBond {
 	// we could directly forward bond sats to the other parties payout address in case it is a taken trade
 	async fn punish(&self, coordinator: &Coordinator) -> Result<()> {
 		// publish bond
+		debug!("Publishing violating bond tx: {}", self.bond_tx_hex);
 		coordinator
 			.coordinator_wallet
 			.publish_bond_tx_hex(&self.bond_tx_hex)?; // can be made async with esplora backend if we figure out the compilation error of bdk
 
 		// remove offer from db/orderbook
+		self.remove_from_db_tables(coordinator.coordinator_db.clone())
+			.await?;
 		Ok(())
 	}
 }

taptrade-cli-demo/coordinator/src/database/mod.rs

@@ -5,7 +5,7 @@ use futures_util::StreamExt;
 
 use super::*;
 use sqlx::{sqlite::SqlitePoolOptions, Pool, Row, Sqlite};
-use std::{collections::HashMap, env};
+use std::env;
 
 #[derive(Clone, Debug)]
 pub struct CoordinatorDB {
@@ -430,52 +430,82 @@ impl CoordinatorDB {
 			bonds.push(bond);
 		}
 
-		let mut rows_taken = sqlx::query(
+		// we shouldn't need this as bonds will be locked onchain when trade is taken and we should
-			"SELECT offer_id, robohash_maker, robohash_taker,
+		// move to taken_offers only once everything is confirmed
-			bond_address_maker, bond_address_taker, bond_amount_sat, amount_sat, bond_tx_hex_maker, bond_tx_hex_taker
+		// let mut rows_taken = sqlx::query(
-			FROM taken_offers",
+		// 	"SELECT offer_id, robohash_maker, robohash_taker,
-		)
+		// 	bond_address_maker, bond_address_taker, bond_amount_sat, amount_sat, bond_tx_hex_maker, bond_tx_hex_taker
-		.fetch(&*self.db_pool);
+		// 	FROM taken_offers",
+		// )
+		// .fetch(&*self.db_pool);
 
-		while let Some(row) = rows_taken.next().await {
+		// while let Some(row) = rows_taken.next().await {
-			let row = row?;
+		// 	let row = row?;
 
-			let robohash_maker: Vec<u8> = row.get("robohash_maker");
+		// 	let robohash_maker: Vec<u8> = row.get("robohash_maker");
-			let robohash_taker: Vec<u8> = row.get("robohash_taker");
+		// 	let robohash_taker: Vec<u8> = row.get("robohash_taker");
-			let locking_amount_sat = row.get::<i64, _>("bond_amount_sat") as u64;
+		// 	let locking_amount_sat = row.get::<i64, _>("bond_amount_sat") as u64;
-			let min_input_sum_sat = row.get::<i64, _>("amount_sat") as u64;
+		// 	let min_input_sum_sat = row.get::<i64, _>("amount_sat") as u64;
-			let trade_id_hex: String = row.get("offer_id");
+		// 	let trade_id_hex: String = row.get("offer_id");
 
-			let requirements_maker = BondRequirements {
+		// 	let requirements_maker = BondRequirements {
-				bond_address: row.get("bond_address_maker"),
+		// 		bond_address: row.get("bond_address_maker"),
-				locking_amount_sat,
+		// 		locking_amount_sat,
-				min_input_sum_sat,
+		// 		min_input_sum_sat,
-			};
+		// 	};
 
-			let bond_maker = MonitoringBond {
+		// 	let bond_maker = MonitoringBond {
-				bond_tx_hex: row.get("bond_tx_hex_maker"),
+		// 		bond_tx_hex: row.get("bond_tx_hex_maker"),
-				robot: robohash_maker,
+		// 		robot: robohash_maker,
-				trade_id_hex: trade_id_hex.clone(),
+		// 		trade_id_hex: trade_id_hex.clone(),
-				requirements: requirements_maker,
+		// 		requirements: requirements_maker,
-				table: Table::ActiveTrades,
+		// 		table: Table::ActiveTrades,
-			};
+		// 	};
-			bonds.push(bond_maker);
+		// 	bonds.push(bond_maker);
 
-			let requirements_maker = BondRequirements {
+		// 	let requirements_maker = BondRequirements {
-				bond_address: row.get("bond_address_taker"),
+		// 		bond_address: row.get("bond_address_taker"),
-				locking_amount_sat,
+		// 		locking_amount_sat,
-				min_input_sum_sat,
+		// 		min_input_sum_sat,
-			};
+		// 	};
 
-			let bond_taker = MonitoringBond {
+		// 	let bond_taker = MonitoringBond {
-				bond_tx_hex: row.get("bond_tx_hex_taker"),
+		// 		bond_tx_hex: row.get("bond_tx_hex_taker"),
-				trade_id_hex,
+		// 		trade_id_hex,
-				robot: robohash_taker,
+		// 		robot: robohash_taker,
-				requirements: requirements_maker,
+		// 		requirements: requirements_maker,
-				table: Table::ActiveTrades,
+		// 		table: Table::ActiveTrades,
-			};
+		// 	};
-			bonds.push(bond_taker);
+		// 	bonds.push(bond_taker);
-		}
+		// }
 		Ok(bonds)
 	}
+
+	pub async fn remove_violating_bond(&self, bond: &MonitoringBond) -> Result<()> {
+		if bond.table == Table::Orderbook {
+			sqlx::query("DELETE FROM active_maker_offers WHERE offer_id = ?")
+				.bind(&bond.trade_id_hex)
+				.execute(&*self.db_pool)
+				.await?;
+			debug!("Removed violating bond offer from orderbook");
+		} else {
+			return Err(anyhow!(
+				"Invalid table type when trying to remove violating bond from db"
+			));
+		}
+
+		// we shouldn't need this as bonds will be locked onchain when trade is taken and we should
+		// move to taken_offers only once everything is confirmed
+		// } else if bond.table == Table::ActiveTrades {
+		// 	sqlx::query("DELETE FROM taken_offers WHERE offer_id = ?")
+		// 		.bind(bond.trade_id_hex)
+		// 		.execute(&*self.db_pool)
+		// 		.await?;
+
+		// sqlx::query("DELETE FROM active_maker_offers WHERE offer_id = ?")
+		// 	.bind(trade_id_hex)
+		// 	.execute(&*self.db_pool)
+		// 	.await?;
+		Ok(())
+	}
 }

taptrade-cli-demo/coordinator/src/main.rs

@@ -13,7 +13,7 @@ use dotenv::dotenv;
 use log::{debug, error, info, warn};
 use std::time::{SystemTime, UNIX_EPOCH};
 use std::{env, sync::Arc};
-use tokio::{sync::Mutex, task::spawn_blocking};
+use tokio::sync::Mutex;
 use wallet::*;
 
 pub struct Coordinator {

taptrade-cli-demo/rpc_node/regtest/data/mine-blocks.sh

@@ -18,8 +18,8 @@ fi
 # Generate initial blocks
 bitcoin-cli -regtest -datadir="/home/bitcoin/.bitcoin" -rpcwallet="coordinator_wallet" -generate 101
 
-# Generate a block every 300 seconds
+# Generate a block every 120 seconds
 while true; do
     bitcoin-cli -regtest -datadir="/home/bitcoin/.bitcoin" -rpcwallet="coordinator_wallet" -generate 1
-    sleep 300
+    sleep 120
 done

taptrade-cli-demo/trader/maker.env

@@ -1,4 +1,4 @@
-ELECTRUM_ENDPOINT="ssl://localhost:50001" # regtest electrum server
+ELECTRUM_ENDPOINT="tcp://localhost:50001" # regtest electrum server
 COORDINATOR_ENDPOINT="http://127.0.0.1:9999"
 ROBOHASH_HEX="26ee3dee4815655d223c3505162fd4610294a9542f89bb3d3e9748f534ac10ae"  # sha256 of "robot21"
 TRADE_TYPE="buy"

taptrade-cli-demo/trader/taker.env

@@ -1,4 +1,4 @@
-ELECTRUM_ENDPOINT="ssl://localhost:50001" # signet electrum server
+ELECTRUM_ENDPOINT="tcp://localhost:50001" # signet electrum server
 COORDINATOR_ENDPOINT="http://127.0.0.1:9999"
 ROBOHASH_HEX="169b6049cf865eba7d01e1ad26975f1d5ff29d570297ff18d40a53c8281dff5d"  # sha256 of "robot22"
 TRADE_TYPE="sell"