Track transactions in the mempool

Cargo.lock

@@ -146,42 +146,12 @@ version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "383d29d513d8764dcdc42ea295d979eb99c3c9f00607b3692cf68a431f7dca72"
 
-[[package]]
-name = "bitcoin"
-version = "0.30.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1945a5048598e4189e239d3f809b19bdad4845c4b2ba400d304d2dcf26d2c462"
-dependencies = [
- "bech32",
- "bitcoin-private",
- "bitcoin_hashes 0.12.0",
- "hex_lit",
- "secp256k1 0.27.0",
- "serde",
-]
-
-[[package]]
-name = "bitcoin-private"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "73290177011694f38ec25e165d0387ab7ea749a4b81cd4c80dae5988229f7a57"
-
 [[package]]
 name = "bitcoin_hashes"
 version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "90064b8dee6815a6470d60bad07bbbaee885c0e12d04177138fa3291a01b7bc4"
 
-[[package]]
-name = "bitcoin_hashes"
-version = "0.12.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d7066118b13d4b20b23645932dfb3a81ce7e29f95726c2036fa33cd7b092501"
-dependencies = [
- "bitcoin-private",
- "serde",
-]
-
 [[package]]
 name = "bitcoincash"
 version = "0.29.2"
@@ -189,8 +159,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dc7b97d0bbf8b62e6cfbec23aa42e677dfced45572a65992c5e08906728d44bf"
 dependencies = [
  "bech32",
- "bitcoin_hashes 0.11.0",
+ "bitcoin_hashes",
- "secp256k1 0.24.3",
+ "secp256k1",
 ]
 
 [[package]]
@@ -234,7 +204,7 @@ name = "cauldron-indexer"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "bitcoin_hashes 0.11.0",
+ "bitcoin_hashes",
  "bitcoincash",
  "electrum-client",
  "hex",
@@ -377,19 +347,15 @@ checksum = "a26ae43d7bcc3b814de94796a5e736d4029efb0ee900c12e2d54c993ad1a1e07"
 
 [[package]]
 name = "electrum-client"
-version = "0.18.0"
+version = "0.19.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
+source = "git+https://gitlab.com/dagurval/rust-electrum-client.git#f0fa7e9c5b1f3a0b7794f4ce1ae477f58ae6ccad"
-checksum = "6bc133f1c8d829d254f013f946653cbeb2b08674b960146361d1e9b67733ad19"
 dependencies = [
- "bitcoin",
- "bitcoin-private",
  "byteorder",
  "libc",
  "log",
  "rustls",
  "serde",
  "serde_json",
- "webpki",
  "webpki-roots",
  "winapi",
 ]
@@ -617,12 +583,6 @@ version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
 
-[[package]]
-name = "hex_lit"
-version = "0.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3011d1213f159867b13cfd6ac92d2cd5f1345762c63be3554e84092d85a50bbd"
-
 [[package]]
 name = "http"
 version = "0.2.11"
@@ -1179,7 +1139,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "582d104183f808e097bbe80149f6a17190f10171e9d0feb9af23763797e51cc6"
 dependencies = [
  "anyhow",
- "bitcoin_hashes 0.11.0",
+ "bitcoin_hashes",
  "bitcoincash",
 ]
 
@@ -1391,19 +1351,8 @@ version = "0.24.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6b1629c9c557ef9b293568b338dddfc8208c98a18c59d722a9d53f859d9c9b62"
 dependencies = [
- "bitcoin_hashes 0.11.0",
+ "bitcoin_hashes",
- "secp256k1-sys 0.6.1",
+ "secp256k1-sys",
-]
-
-[[package]]
-name = "secp256k1"
-version = "0.27.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "25996b82292a7a57ed3508f052cfff8640d38d32018784acd714758b43da9c8f"
-dependencies = [
- "bitcoin_hashes 0.12.0",
- "secp256k1-sys 0.8.1",
- "serde",
 ]
 
 [[package]]
@@ -1415,15 +1364,6 @@ dependencies = [
  "cc",
 ]
 
-[[package]]
-name = "secp256k1-sys"
-version = "0.8.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70a129b9e9efbfb223753b9163c4ab3b13cff7fd9c7f010fbac25ab4099fa07e"
-dependencies = [
- "cc",
-]
-
 [[package]]
 name = "serde"
 version = "1.0.193"
@@ -1971,24 +1911,11 @@ version = "0.2.91"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4f186bd2dcf04330886ce82d6f33dd75a7bfcf69ecf5763b89fcde53b6ac9838"
 
-[[package]]
-name = "webpki"
-version = "0.22.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ed63aea5ce73d0ff405984102c42de94fc55a6b75765d621c65262469b3c9b53"
-dependencies = [
- "ring",
- "untrusted",
-]
-
 [[package]]
 name = "webpki-roots"
-version = "0.22.6"
+version = "0.25.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6c71e40d7d2c34a5106301fb632274ca37242cd0c9d3e64dbece371a40a2d87"
+checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1"
-dependencies = [
- "webpki",
-]
 
 [[package]]
 name = "winapi"

Cargo.toml

@@ -14,7 +14,7 @@ readme = "README.md"
 anyhow = "1.0.75"
 bitcoincash = "0.29.2"
 bitcoin_hashes = { version = "0.11.0", default-features = false } # Same version as used by bitcoincash
-electrum-client = "0.18.0"
+electrum-client = { git = "https://gitlab.com/dagurval/rust-electrum-client.git" }
 hex = "0.4.3"
 serde_json = "1.0.108"
 rocket = { version = "0.5.0", features = ["json"] }

src/chain.rs

@@ -296,7 +296,10 @@ pub fn download_all_headers(electrum: &Client) -> Result<Vec<NewHeader>> {
         info!("Fetching headers {} -> {}", batch_start, batch_end);
 
         for height in batch_start..=batch_end {
-            batch.block_header(height as u32);
+            batch.raw(
+                "blockchain.block.header".to_owned(),
+                vec![Param::U32(height as u32)],
+            );
         }
 
         let headers_response = electrum.batch_call(&batch)?;

src/db.rs

@@ -3,11 +3,19 @@
 // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
 // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
 
-use anyhow::{bail, Result};
+use std::{
-use bitcoin_hashes::hex::ToHex;
+    collections::HashSet,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use anyhow::{bail, Context, Result};
+use bitcoin_hashes::{
+    hex::{FromHex, ToHex},
+    Hash,
+};
 use bitcoincash::{
     consensus::{deserialize, serialize},
-    BlockHash, BlockHeader,
+    BlockHash, BlockHeader, Txid,
 };
 use riftenlabs_defi::cauldron::ParsedContract;
 use rusqlite::{params, Connection, Transaction};
@@ -18,6 +26,7 @@ pub fn prepare_tables(conn: &Connection) {
     conn.execute(
         "CREATE TABLE utxo_funding (
         new_utxo_hash TEXT PRIMARY KEY,
+        txid TEXT,
         spent_utxo_hash TEXT,
         timestamp BIGINT,
         new_utxo_txid TEXT,
@@ -43,6 +52,7 @@ pub fn prepare_tables(conn: &Connection) {
     conn.execute(
         "CREATE TABLE utxo_spending (
         spent_utxo_hash TEXT PRIMARY KEY,
+        txid TEXT,
         timestamp BIGINT,
         blockhash TEXT
     );",
@@ -73,6 +83,15 @@ pub fn prepare_tables(conn: &Connection) {
     )
     .unwrap();
 
+    conn.execute(
+        "CREATE TABLE first_seen (
+        txid TEXT PRIMARY KEY,
+        timestamp INT
+    )",
+        [],
+    )
+    .unwrap();
+
     // Indexes for list_by_volume
     conn.execute(
         "CREATE INDEX idx_utxo_funding_join ON utxo_funding(new_utxo_hash, sats, token_id);",
@@ -111,9 +130,16 @@ pub fn insert_utxo_funding(
     timestamp: u32,
     cauldrons: &Vec<ParsedContract>,
     blockhash: &BlockHash,
+    txid: &Txid,
+    replace: bool,
 ) -> Result<()> {
     let mut statement = con
-        .prepare("INSERT OR IGNORE INTO utxo_funding (new_utxo_hash, spent_utxo_hash, timestamp, new_utxo_txid, new_utxo_n, sats, token_amount, token_id, blockhash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")?;
+        .prepare(
+            &format!("INSERT OR {} INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, timestamp, new_utxo_txid, new_utxo_n, sats, token_amount, token_id, blockhash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            if replace {
+                "REPLACE"
+            } else { "IGNORE"}
+        ))?;
 
     for c in cauldrons {
         if c.new_utxo_hash.is_none() {
@@ -121,6 +147,7 @@ pub fn insert_utxo_funding(
         }
         statement.execute(params![
             c.new_utxo_hash.unwrap().to_hex(),
+            txid.to_hex(),
             c.spent_utxo_hash.to_hex(),
             timestamp,
             c.new_utxo_txid.unwrap().to_hex(),
@@ -139,14 +166,17 @@ pub fn insert_utxo_spending(
     timestamp: u32,
     cauldrons: &Vec<ParsedContract>,
     blockhash: &BlockHash,
+    txid: &Txid,
+    replace: bool,
 ) -> Result<()> {
     let mut statement = tx.prepare(
-        "INSERT OR IGNORE INTO utxo_spending (spent_utxo_hash, timestamp, blockhash) VALUES (?, ?, ?)",
+        &format!("INSERT OR {} INTO utxo_spending (spent_utxo_hash, txid, timestamp, blockhash) VALUES (?, ?, ?, ?)", if replace { "REPLACE" } else { "IGNORE "}),
     )?;
 
     for c in cauldrons {
         statement.execute(params![
             c.spent_utxo_hash.to_hex(),
+            txid.to_hex(),
             timestamp,
             blockhash.to_hex()
         ])?;
@@ -168,9 +198,11 @@ pub fn get_token_tvl(
                 utxo_funding uf
             LEFT JOIN
                 utxo_spending us ON uf.new_utxo_hash = us.spent_utxo_hash
+            LEFT JOIN
+                first_seen fs ON uf.txid = fs.txid
             WHERE
-                (us.timestamp IS NULL OR us.timestamp > ?)
+                (us.timestamp IS NULL OR us.timestamp > ? OR us.timestamp = 0) -- unspent or spent after max timestamp -- or in mempool
-                AND uf.timestamp <= ?
+                AND COALESCE(fs.timestamp, uf.timestamp) <= ? -- filter out anything above max timestamp
             GROUP BY
                 uf.token_id",
     )?;
@@ -205,7 +237,8 @@ pub fn list_tokens_by_volume(
                     ABS(uf1.sats - COALESCE(uf2.sats, 0)) as trade_volume
                 FROM utxo_funding uf1
                 LEFT JOIN utxo_funding uf2 ON uf1.spent_utxo_hash = uf2.new_utxo_hash
-                WHERE uf1.timestamp >= (strftime('%s', 'now') - ?)
+                LEFT JOIN first_seen fs1 ON uf1.txid = fs1.txid
+                WHERE COALESCE(fs1.timestamp, uf1.timestamp) >= (strftime('%s', 'now') - ?)
             ),
             TVLData AS (
                 SELECT
@@ -371,17 +404,19 @@ pub fn historic_price(
     // Prepare and execute the SQL query for the current interval
     let sql = "
     SELECT
-        utxo_funding.timestamp,
+        COALESCE(first_seen.timestamp, utxo_funding.timestamp) AS effective_timestamp,
         utxo_funding.sats,
         utxo_funding.token_amount
     FROM
         utxo_funding
+    LEFT JOIN
+        first_seen ON utxo_funding.txid = first_seen.txid
     WHERE
         utxo_funding.token_id = ? AND
-        utxo_funding.timestamp >= ? AND
+        effective_timestamp >= ? AND
-        utxo_funding.timestamp < ?
+        effective_timestamp < ?
     ORDER BY
-        utxo_funding.timestamp ASC
+        effective_timestamp ASC
 ";
     let mut statement = connection.prepare(sql)?;
 
@@ -441,7 +476,7 @@ pub fn load_all_headers(conn: &Connection) -> Result<Vec<(BlockHeader, u64)>> {
 }
 
 /// Store headers to database
-pub fn store_headers(conn: &Connection, headers: &Vec<NewHeader>) -> Result<()> {
+pub fn store_headers(conn: &Connection, headers: &[NewHeader]) -> Result<()> {
     let mut stmt =
         conn.prepare("INSERT OR IGNORE INTO headers (key, height, header) VALUES (?, ?, ?)")?;
 
@@ -451,3 +486,45 @@ pub fn store_headers(conn: &Connection, headers: &Vec<NewHeader>) -> Result<()>
 
     Ok(())
 }
+
+/// Load all headers we have stored in our database
+pub fn db_load_mempool(conn: &Connection) -> Result<HashSet<Txid>> {
+    let mut stmt = conn.prepare("SELECT txid FROM utxo_funding WHERE blockhash = ?")?;
+    let blockhash = BlockHash::all_zeros().to_hex();
+    let txid_iter = stmt.query_map(params![blockhash], |row| row.get(0))?;
+
+    let mut txids: HashSet<Txid> = HashSet::new();
+    for txid_res in txid_iter {
+        let txid_hex: String = txid_res?;
+        let txid = Txid::from_hex(&txid_hex).expect("invalid txid in db");
+        txids.insert(txid);
+    }
+    Ok(txids)
+}
+
+pub fn db_delete_mempool_tx(db_tx: &Connection, txid: &Txid) -> Result<bool> {
+    let blockhash = BlockHash::all_zeros();
+
+    let mut stmt = db_tx.prepare("DELETE FROM utxo_spending WHERE blockhash = ? AND txid = ?")?;
+    let rows_deleted_spending = stmt.execute(params![blockhash.to_hex(), txid.to_hex()])?;
+
+    let mut stmt = db_tx.prepare("DELETE FROM utxo_funding WHERE blockhash = ? AND txid = ?")?;
+    let rows_deleted_funding = stmt.execute(params![blockhash.to_hex(), txid.to_hex()])?;
+
+    Ok(rows_deleted_spending != 0 || rows_deleted_funding != 0)
+}
+
+pub fn db_add_first_seen(db_tx: &Connection, txid: &Txid) -> Result<()> {
+    let mut stmt = db_tx
+        .prepare("INSERT OR IGNORE INTO first_seen (txid, timestamp) VALUES (?, ?)")
+        .context("add_first_seen stmpt error")?;
+
+    let current_timestamp = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64;
+
+    stmt.execute(params![txid.to_hex(), current_timestamp])?;
+
+    Ok(())
+}

src/electrum.rs

@@ -3,10 +3,15 @@
 // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
 // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
 
+use std::{collections::HashSet, str::FromStr};
+
 use anyhow::{Context, Result};
-use bitcoincash::{consensus::deserialize, BlockHeader};
+use bitcoin_hashes::hex::ToHex;
-use electrum_client::{Client, ElectrumApi};
+use bitcoincash::{consensus::deserialize, BlockHeader, Transaction, Txid};
-use serde_json::Value;
+use electrum_client::{Client, ElectrumApi, Param};
+use log::info;
+use riftenlabs_defi::cauldron::V2_CONTRACT_TEMPLATE;
+use serde_json::{json, Value};
 
 /// Fetch blockchain tip from electrum server
 pub fn electrum_get_tip(client: &Client) -> Result<(BlockHeader, u64)> {
@@ -27,3 +32,49 @@ pub fn electrum_get_tip(client: &Client) -> Result<(BlockHeader, u64)> {
 
     Ok((deserialize(&hex::decode(header)?)?, height as u64))
 }
+
+/// Fetch cauldron mempool transactions
+pub fn electrum_fetch_mempool(client: &Client) -> Result<HashSet<Txid>> {
+    let filter = json!({
+        "scriptsig": hex::encode(&V2_CONTRACT_TEMPLATE[(V2_CONTRACT_TEMPLATE.len() - 43)..])
+    });
+
+    let response = client.raw_call("mempool.get", [Param::Value(filter)])?;
+
+    let txs = response
+        .get("transactions")
+        .context("no txs in mempool get")?
+        .as_array()
+        .context("txs not array")?;
+    Ok(txs
+        .iter()
+        .filter_map(|txid| match txid.as_str() {
+            Some(txid_hex) => match Txid::from_str(txid_hex) {
+                Ok(txid) => Some(txid),
+                Err(e) => {
+                    info!("Txid not hex: {}", e);
+                    None
+                }
+            },
+            None => {
+                info!("Failed to read txid from electrum response {:?}", txid);
+                None
+            }
+        })
+        .collect())
+}
+
+/// Fetch blockchain tip from electrum server
+pub fn electrum_get_tx(client: &Client, txid: &Txid) -> Result<Transaction> {
+    let tx: Value = serde_json::from_str(
+        &client
+            .raw_call("blockchain.transaction.get", [Param::String(txid.to_hex())])?
+            .to_string(),
+    )?;
+
+    let tx: Transaction = deserialize(
+        &hex::decode(tx.as_str().context("no tx in response")?).context("failed to decode tx")?,
+    )?;
+
+    Ok(tx)
+}

src/main.rs

@@ -4,11 +4,17 @@
 // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
 
 use anyhow::Result;
-use bitcoin_hashes::hex::{FromHex, ToHex};
+use bitcoin_hashes::{
-use bitcoincash::{consensus::deserialize, Block, BlockHash};
+    hex::{FromHex, ToHex},
+    Hash,
+};
+use bitcoincash::{consensus::deserialize, Block, BlockHash, Transaction, Txid};
 use chain::{get_new_headers, Chain, StoreBlockUndoer};
-use db::{get_token_tvl, historic_price, list_tokens_by_volume, load_all_headers, store_headers};
+use db::{
-use electrum::electrum_get_tip;
+    db_load_mempool, get_token_tvl, historic_price, list_tokens_by_volume, load_all_headers,
+    store_headers,
+};
+use electrum::{electrum_fetch_mempool, electrum_get_tip};
 use electrum_client::{Client, ElectrumApi, Param};
 use log::{debug, error, info};
 use rayon::prelude::*;
@@ -30,6 +36,7 @@ use rusqlite::Connection;
 use serde_json::Value;
 use std::{
     backtrace::Backtrace,
+    collections::HashSet,
     panic,
     path::Path,
     process,
@@ -39,8 +46,12 @@ use std::{
 };
 use stderrlog::LogLevelNum;
 
-use crate::db::{
+use crate::{
-    config_get, config_set, insert_utxo_funding, insert_utxo_spending, prepare_tables,
+    db::{
+        config_get, config_set, db_add_first_seen, db_delete_mempool_tx, insert_utxo_funding,
+        insert_utxo_spending, prepare_tables,
+    },
+    electrum::electrum_get_tx,
 };
 
 // The block where first cauldron contract was deployed. (Block 799870)
@@ -54,19 +65,62 @@ mod chain;
 mod db;
 mod electrum;
 
-fn parse_cauldrons(block: &Block) -> Vec<ParsedContract> {
+fn parse_cauldrons(tx: &Transaction) -> Vec<ParsedContract> {
-    block
+    tx.input
-        .txdata
         .par_iter()
-        .flat_map(|tx| {
+        .enumerate()
-            tx.input
+        .filter_map(move |(i, _)| parse_cauldron(i, tx))
-                .par_iter()
-                .enumerate()
-                .filter_map(move |(i, _)| parse_cauldron(i, tx))
-        })
         .collect()
 }
 
+fn update_mempool(db: Arc<Mutex<Connection>>, electrum: Arc<Mutex<Client>>) -> Result<()> {
+    let our_mempool_txs: HashSet<Txid> = db_load_mempool(&db.lock().unwrap())?;
+    let node_mempool: HashSet<Txid> = electrum_fetch_mempool(&electrum.lock().unwrap())?;
+
+    let txs_to_delete = our_mempool_txs.difference(&node_mempool);
+    let txs_to_add = node_mempool.difference(&our_mempool_txs);
+
+    let txs_to_add: Vec<Transaction> = txs_to_add
+        .into_iter()
+        .filter_map(
+            |txid| match electrum_get_tx(&electrum.lock().unwrap(), txid) {
+                Ok(tx) => Some(tx),
+                Err(e) => {
+                    info!("Failed to get mempool tx {}: {}", txid, e);
+                    None
+                }
+            },
+        )
+        .collect();
+
+    let mut db_lock = db.lock().unwrap();
+    let db_tx = db_lock.transaction()?;
+
+    for txid in txs_to_delete {
+        debug!("mempool remove {}", txid.to_hex());
+        db_delete_mempool_tx(&db_tx, txid)?;
+    }
+
+    for tx in txs_to_add {
+        let txid = tx.txid();
+        debug!("mempool add {}", txid.to_hex());
+        db_add_first_seen(&db_tx, &txid)?;
+        let cauldrons: Vec<ParsedContract> = tx
+            .input
+            .iter()
+            .enumerate()
+            .filter_map(|(i, _)| parse_cauldron(i, &tx))
+            .collect();
+
+        let mtp = 0;
+        let blockhash = BlockHash::all_zeros();
+        insert_utxo_funding(&db_tx, mtp, &cauldrons, &blockhash, &txid, false)?;
+        insert_utxo_spending(&db_tx, mtp, &cauldrons, &blockhash, &txid, false)?;
+    }
+
+    Ok(db_tx.commit()?)
+}
+
 fn index_blocks(
     chain: Arc<Mutex<Chain>>,
     conn: Arc<Mutex<Connection>>,
@@ -88,7 +142,14 @@ fn index_blocks(
             let new_headers =
                 get_new_headers(&client.lock().unwrap(), &chain, &tip_header.block_hash())?;
             debug!("Storing headers");
-            store_headers(&conn.lock().unwrap(), &new_headers)?;
+            {
+                let mut db_lock = conn.lock().unwrap();
+                for chunk in new_headers.chunks(100000) {
+                    let db_tx = db_lock.transaction()?;
+                    store_headers(&db_tx, chunk)?;
+                    db_tx.commit()?;
+                }
+            }
             let undoer = StoreBlockUndoer::new(conn.clone())?;
             chain.update(undoer, new_headers, None)?;
             debug!("Header update done");
@@ -146,20 +207,29 @@ fn index_blocks(
         };
 
         let mut lock = conn.lock().unwrap();
-        let tx = lock.transaction()?;
+        let db_tx = lock.transaction()?;
-        let cauldrons = parse_cauldrons(&block);
+
         let blockhash = block.block_hash();
-        insert_utxo_funding(&tx, mtp as u32, &cauldrons, &blockhash)?;
+
-        insert_utxo_spending(&tx, mtp as u32, &cauldrons, &blockhash)?;
+        let mut total_cauldrons = 0;
-        config_set(&tx, KEY_LAST_INDEXED, &blockhash.to_hex());
+
-        tx.commit()?;
+        for tx in block.txdata {
+            let cauldrons = parse_cauldrons(&tx);
+            let txid = tx.txid();
+            insert_utxo_funding(&db_tx, mtp as u32, &cauldrons, &blockhash, &txid, true)?;
+            insert_utxo_spending(&db_tx, mtp as u32, &cauldrons, &blockhash, &txid, true)?;
+            total_cauldrons += cauldrons.len();
+        }
+
+        config_set(&db_tx, KEY_LAST_INDEXED, &blockhash.to_hex());
+        db_tx.commit()?;
 
         println!(
             "Indexed {}; mtp: {}, height {}, {} trades.",
             block.header.block_hash().to_hex(),
             mtp,
             block_height,
-            cauldrons.len()
+            total_cauldrons
         );
     }
     Ok(tip_header.block_hash())
@@ -346,7 +416,10 @@ fn start_program() -> Result<Arc<Mutex<Connection>>> {
                     }
                 }
             }
-            thread::sleep(Duration::from_secs(60));
+            if let Err(e) = update_mempool(conn.clone(), client.clone()) {
+                error!("Failed to update mempool: {}", e);
+            }
+            thread::sleep(Duration::from_secs(5));
         }
     });