Correct tracker mempool schema and refine the filtering

Author: Wukong247

src/db/db_manager.rs

@@ -54,14 +54,17 @@ pub async fn run(
             DbRequest::WatchUtxo(outpoint, resp_tx) => {
                 info!("Watch utxo intercepted");
 
-                let mempool_tx = mempool_tx::table
+                let mut mempool_tx = mempool_tx::table
                     .inner_join(mempool_inputs::table.on(mempool_tx::txid.eq(mempool_inputs::txid)))
                     .filter(mempool_inputs::input_txid.eq(outpoint.txid.to_string()))
                     .filter(mempool_inputs::input_vout.eq(outpoint.vout as i32))
                     .select((mempool_tx::txid, mempool_tx::seen_at))
                     .load::<MempoolTx>(&mut conn)
                     .unwrap();
 
+                mempool_tx.sort_by(|a, b| a.txid.cmp(&b.txid));
+                mempool_tx.dedup_by(|a, b| a.txid == b.txid);
+
                 let _ = resp_tx.send(mempool_tx).await;
             }
         }

src/db/schema.rs

@@ -1,8 +1,7 @@
 // @generated automatically by Diesel CLI.
 
 diesel::table! {
-    mempool_inputs (rowid) {
-        rowid -> Integer,
+    mempool_inputs (txid, input_txid, input_vout) {
         txid -> Text,
         input_txid -> Text,
         input_vout -> Integer,

src/indexer/tracker_indexer.rs

@@ -61,7 +61,10 @@ pub async fn run(
             }
         }
         last_tip = tip_height;
+        #[cfg(not(feature = "integration-test"))]
         tokio::time::sleep(Duration::from_secs(10)).await;
+        #[cfg(feature = "integration-test")]
+        tokio::time::sleep(Duration::from_millis(100)).await;
     }
 }
 

src/indexer/utxo_indexer.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use crate::db::model::{MempoolInput, MempoolTx, Utxo};
 use crate::db::schema::{mempool_inputs, mempool_tx, utxos};
 use crate::indexer::rpc::BitcoinRpc;
-use chrono::NaiveDateTime;
+use chrono::{NaiveDateTime, Utc};
 use diesel::SqliteConnection;
 use diesel::prelude::*;
 use diesel::r2d2::ConnectionManager;
@@ -30,7 +30,7 @@ impl<'a> Indexer<'a> {
 
             for input in &tx.input {
                 let prevout = &input.previous_output;
-                diesel::insert_into(mempool_inputs::table)
+                diesel::insert_or_ignore_into(mempool_inputs::table)
                     .values(&MempoolInput {
                         txid: txid.to_string(),
                         input_txid: prevout.txid.to_string(),
@@ -40,7 +40,7 @@ impl<'a> Indexer<'a> {
                     .unwrap();
 
                 self.mark_utxo_spent(
-                    &txid.to_string(),
+                    &prevout.txid.to_string(),
                     prevout.vout as i32,
                     Some(&txid.to_string()),
                     false,
@@ -106,7 +106,7 @@ impl<'a> Indexer<'a> {
         diesel::insert_or_ignore_into(mempool_tx::table)
             .values(&MempoolTx {
                 txid: txid.to_string(),
-                seen_at: NaiveDateTime::MIN,
+                seen_at: Utc::now().naive_utc(),
             })
             .execute(&mut conn)
             .unwrap();

src/server/tracker_monitor.rs

@@ -21,7 +21,7 @@ use crate::{
 
 use tokio::io::BufReader;
 
-const COOLDOWN_PERIOD: u64 = 5;
+const COOLDOWN_PERIOD: u64 = 15 * 60;
 pub async fn monitor_systems(
     db_tx: Sender<DbRequest>,
     status_tx: status::Sender,
@@ -73,7 +73,9 @@ pub async fn monitor_systems(
                             };
                             _ = send_message_with_prefix(&mut writer, &message).await;
 
-                            let buffer = handle_result!(status_tx, read_message(&mut reader).await);
+                            let Ok(buffer) = read_message(&mut reader).await else {
+                                continue;
+                            };
                             let response: TrackerClientToServer =
                                 match serde_cbor::de::from_reader(&buffer[..]) {
                                     Ok(resp) => resp,
@@ -114,7 +116,7 @@ pub async fn monitor_systems(
                     let _ = db_tx.send(DbRequest::Update(address, updated_info)).await;
                 }
             }
-            sleep(Duration::from_secs(4)).await;
+            sleep(Duration::from_secs(COOLDOWN_PERIOD)).await;
         }
     }
 }

src/server/tracker_server.rs

@@ -1,4 +1,5 @@
 use crate::db::model::MempoolTx;
+use crate::server::send_message_with_prefix;
 use crate::server::tracker_monitor::monitor_systems;
 use crate::status;
 use crate::types::DbRequest;
@@ -120,7 +121,7 @@ async fn handle_client(mut stream: TcpStream, db_tx: Sender<DbRequest>) {
 
                 if let Some(mempool_tx) = response {
                     let message = TrackerServerToClient::WatchResponse { mempool_tx };
-                    if let Err(e) = send_message(&mut writer, &message).await {
+                    if let Err(e) = send_message_with_prefix(&mut writer, &message).await {
                         error!("Failed to send response to client: {e}");
                         break;
                     }