From c3f3c269e2e00a2cd6caf0f207c1d75e333008ad Mon Sep 17 00:00:00 2001 From: Hossein Zoda Date: Mon, 6 Jul 2026 08:59:46 +0000 Subject: [PATCH] bcmr: index new registrations from mempool, no confirmation needed Add an OP_RETURN "BCMR" prefix filter to the rostrum mempool.get pass and index matching txs in update_mempool with the all-zeros sentinel blockhash (same pattern as oracle). A newly registered BCMR becomes the auth head and is downloaded immediately; the confirming block re-stamps the entry with its real blockhash in place. - insert_authheader: INSERT OR REPLACE -> upsert. REPLACE deletes the row, cascading away downloaded bcmr_data on every mempool->confirmed upgrade. - update_mempool drops sentinel entries whose tx left the mempool (evicted or replaced), so stale auth heads never linger. - clear bcmr mempool state at startup; always-run migration adds txid and blockhash indexes on auth_chain_entry for the per-pass lookups. Auth chain transfers without a BCMR output still index at confirmation. Co-Authored-By: Claude Fable 5 --- src/db/bcmr/mod.rs | 196 ++++++++++++++++++++++++++++++++++++++++++++- src/db/init.rs | 2 + src/electrum.rs | 15 +++- src/index.rs | 43 +++++++++- src/main.rs | 1 + 5 files changed, 251 insertions(+), 6 deletions(-) diff --git a/src/db/bcmr/mod.rs b/src/db/bcmr/mod.rs index 4462538..eab610c 100644 --- a/src/db/bcmr/mod.rs +++ b/src/db/bcmr/mod.rs @@ -7,6 +7,7 @@ use crate::bcmr::parsedbcmr::{FileMeta, ParsedBCMR, Token, Uris, SOURCE_ON_CHAIN use crate::db::blob::{display_hex_to_blob, FromBlob, ToBlob}; use anyhow::*; +use bitcoin_hashes::Hash; use bitcoincash::{BlockHash, TokenID, Txid}; use log::info; use riftenlabs_defi::chainutil::OutPointHash; @@ -116,6 +117,23 @@ pub async fn prepare_tables(pool: &SqlitePool) { .execute(pool) .await .expect("failed to create index"); + + ensure_indexes(pool).await; +} + +/// Always-run migration: indexes for the txid and blockhash lookups used by +/// mempool indexing. Safe on both new and existing bcmr.db. +pub async fn ensure_indexes(pool: &SqlitePool) { + sqlx::query("CREATE INDEX IF NOT EXISTS idx_auth_chain_txid ON auth_chain_entry(txid)") + .execute(pool) + .await + .expect("failed to create index"); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_auth_chain_blockhash ON auth_chain_entry(blockhash)", + ) + .execute(pool) + .await + .expect("failed to create index"); } /// Delete all well-known BCMR entries for the given source. @@ -131,6 +149,51 @@ pub async fn delete_entries_for_well_known<'e>( Ok(()) } +/// Whether any auth chain entry (confirmed or mempool) was indexed from this tx. +pub async fn has_indexed_tx(pool: &SqlitePool, txid: &Txid) -> Result { + let row: Option<(i64,)> = + sqlx::query_as("SELECT 1 FROM auth_chain_entry WHERE txid = ? LIMIT 1") + .bind(txid.to_blob()) + .fetch_optional(pool) + .await?; + Ok(row.is_some()) +} + +/// Txids of auth chain entries indexed from the mempool (all-zeros blockhash sentinel). +pub async fn get_unconfirmed_txids(pool: &SqlitePool) -> Result> { + let rows = sqlx::query("SELECT DISTINCT txid FROM auth_chain_entry WHERE blockhash = ?") + .bind(BlockHash::all_zeros().to_blob()) + .fetch_all(pool) + .await?; + + let mut txids = Vec::with_capacity(rows.len()); + for r in rows { + let txid_blob: Vec = r.get(0); + txids.push(Txid::from_blob(&txid_blob).context("failed to decode txid blob")?); + } + Ok(txids) +} + +/// Delete a mempool-indexed (all-zeros blockhash) auth chain entry. Entries the +/// tx confirmed into are untouched: they carry the real blockhash by then. +pub async fn delete_unconfirmed_tx(pool: &SqlitePool, txid: &Txid) -> Result<()> { + sqlx::query("DELETE FROM auth_chain_entry WHERE blockhash = ? AND txid = ?") + .bind(BlockHash::all_zeros().to_blob()) + .bind(txid.to_blob()) + .execute(pool) + .await?; + Ok(()) +} + +/// Delete all mempool-indexed (all-zeros blockhash) auth chain entries. +pub async fn clear_mempool(pool: &SqlitePool) -> Result { + let r = sqlx::query("DELETE FROM auth_chain_entry WHERE blockhash = ?") + .bind(BlockHash::all_zeros().to_blob()) + .execute(pool) + .await?; + Ok(r.rows_affected() as usize) +} + pub async fn delete_entries_for_block(pool: &SqlitePool, blockhash: &BlockHash) -> Result { let r = sqlx::query("DELETE FROM auth_chain_entry WHERE blockhash = ?") .bind(blockhash.to_blob()) @@ -149,10 +212,18 @@ pub async fn insert_authheader( height: usize, bcmr_data: Option>, ) -> Result<()> { + // Upsert rather than INSERT OR REPLACE: REPLACE deletes the existing row, + // which cascades into bcmr_data and would throw away already-downloaded + // metadata every time a mempool-indexed entry is re-indexed on confirmation. sqlx::query( - "INSERT OR REPLACE INTO auth_chain_entry + "INSERT INTO auth_chain_entry (utxo, blockhash, txid, token_id, height, bcmr_data) - VALUES (?, ?, ?, ?, ?, ?)", + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(token_id, utxo) DO UPDATE SET + blockhash = excluded.blockhash, + txid = excluded.txid, + height = excluded.height, + bcmr_data = excluded.bcmr_data", ) .bind(utxo.to_blob()) .bind(blockhash.to_blob()) @@ -506,6 +577,127 @@ pub async fn filter_tokens_with_bcmr( mod tests { use super::*; + #[tokio::test] + async fn confirming_mempool_entry_preserves_downloaded_bcmr_data() { + use sqlx::sqlite::SqliteConnectOptions; + use std::str::FromStr; + + // foreign_keys on, as in production (db/init.rs): with INSERT OR + // REPLACE the confirmation re-insert would delete the row and cascade + // away bcmr_data; the upsert must keep it. + let opts = SqliteConnectOptions::from_str("sqlite::memory:") + .unwrap() + .foreign_keys(true); + let pool = SqlitePool::connect_with(opts).await.unwrap(); + prepare_tables(&pool).await; + + let token_id = "0101010101010101010101010101010101010101010101010101010101010101" + .parse::() + .unwrap(); + let txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + .parse::() + .unwrap(); + let utxo = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + .parse::() + .unwrap(); + + // Indexed from mempool: sentinel blockhash + insert_authheader( + &pool, + &utxo, + &BlockHash::all_zeros(), + &txid, + &token_id, + 0, + Some(vec![0x6a]), + ) + .await + .unwrap(); + assert_eq!(get_unconfirmed_txids(&pool).await.unwrap(), vec![txid]); + + // Downloader stores the metadata while the tx is still unconfirmed + let parsed = ParsedBCMR { + name: "Test".into(), + description: "".into(), + token: Token { + category: token_id.to_string(), + symbol: "TST".into(), + decimals: 0, + }, + uris: Uris { + web: None, + icon: None, + }, + filemeta: FileMeta { + expected_hash: None, + actual_hash: None, + source: SOURCE_ON_CHAIN.into(), + }, + }; + insert_bcmr_data(&pool, &token_id, &utxo, &parsed) + .await + .unwrap(); + + // Tx confirms: same entry re-indexed with the real blockhash + let real_blockhash = "1111111111111111111111111111111111111111111111111111111111111111" + .parse::() + .unwrap(); + insert_authheader( + &pool, + &utxo, + &real_blockhash, + &txid, + &token_id, + 0, + Some(vec![0x6a]), + ) + .await + .unwrap(); + + assert!(get_unconfirmed_txids(&pool).await.unwrap().is_empty()); + + let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM bcmr_data") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count.0, 1, "downloaded bcmr_data must survive confirmation"); + + let bh: (Vec,) = sqlx::query_as("SELECT blockhash FROM auth_chain_entry") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(bh.0, real_blockhash.to_blob()); + } + + #[tokio::test] + async fn delete_unconfirmed_tx_leaves_confirmed_entries() { + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); + prepare_tables(&pool).await; + + let token_id = "0101010101010101010101010101010101010101010101010101010101010101" + .parse::() + .unwrap(); + let txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + .parse::() + .unwrap(); + let utxo = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + .parse::() + .unwrap(); + let real_blockhash = "1111111111111111111111111111111111111111111111111111111111111111" + .parse::() + .unwrap(); + + insert_authheader(&pool, &utxo, &real_blockhash, &txid, &token_id, 0, None) + .await + .unwrap(); + + // A confirmed entry from the same tx must not be dropped by the + // mempool cleanup pass. + delete_unconfirmed_tx(&pool, &txid).await.unwrap(); + + assert!(has_indexed_tx(&pool, &txid).await.unwrap()); + } + #[tokio::test] async fn test_update_bcmr_failure_multiple_tokens_same_utxo_txid() { let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); diff --git a/src/db/init.rs b/src/db/init.rs index 4d93eb4..440b955 100644 --- a/src/db/init.rs +++ b/src/db/init.rs @@ -115,6 +115,8 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul if !db_exists { bcmr_prepare_tables(&bcmr_db_write).await; } + // Always-run migration: safe on both new and existing bcmr.db + crate::db::bcmr::ensure_indexes(&bcmr_db_write).await; // Initialize CRC20 database let (db_exists, crc20_db_write, crc20_db_read) = diff --git a/src/electrum.rs b/src/electrum.rs index e044e8d..0ed3853 100644 --- a/src/electrum.rs +++ b/src/electrum.rs @@ -18,6 +18,7 @@ use riftenlabs_defi::{ }; use serde_json::{json, Value}; +use crate::bcmr::BCMR_PREFIX; use crate::db::ido::{IDO_PREINIT_ANNOUNCEMENT_SIGNATURE, IDO_SIGNATURE}; /// Fetch blockchain tip from electrum server @@ -40,10 +41,10 @@ pub fn electrum_get_tip(client: &Client) -> Result<(BlockHeader, u64)> { Ok((deserialize(&hex::decode(header)?)?, height as u64)) } -/// Fetch defi (cauldron + tokentoken) and oracle mempool transactions +/// Fetch defi (cauldron + tokentoken), oracle, ido and bcmr mempool transactions pub fn electrum_fetch_mempool( client: &Client, -) -> Result<(HashSet, HashSet, HashSet)> { +) -> Result<(HashSet, HashSet, HashSet, HashSet)> { let cauldron_filter = json!({ "scriptsig": hex::encode(&V2_CONTRACT_TEMPLATE[(V2_CONTRACT_TEMPLATE.len() - 43)..]), // cauldron spends "scriptpubkey": hex::encode([0x6a /* op_return */, 0x06 /* push */, b'S', b'U', b'M', b'M', b'O', b'N']), // new pools (potentially) @@ -82,6 +83,13 @@ pub fn electrum_fetch_mempool( "operation": "union" }); + // BCMR registrations carry an `OP_RETURN "BCMR"` output (genesis or auth + // chain update). Auth chain transfers without a BCMR output are only + // picked up once confirmed. + let bcmr_filter = json!({ + "scriptpubkey": hex::encode(BCMR_PREFIX), + }); + let fetch_txs = |filter: Value| -> Result> { let response = client.raw_call("mempool.get", [Param::Value(filter)])?; @@ -115,8 +123,9 @@ pub fn electrum_fetch_mempool( let mut oracle_txs = fetch_txs(oracle_v1_filter)?; oracle_txs.extend(fetch_txs(oracle_v2_filter)?); let ido_txs = fetch_txs(ido_filter)?; + let bcmr_txs = fetch_txs(bcmr_filter)?; - Ok((defi_txs, oracle_txs, ido_txs)) + Ok((defi_txs, oracle_txs, ido_txs, bcmr_txs)) } /// Fetch blockchain tip from electrum server diff --git a/src/index.rs b/src/index.rs index 4ec99e7..1fa66e0 100644 --- a/src/index.rs +++ b/src/index.rs @@ -77,7 +77,7 @@ pub async fn update_mempool( db::cauldron::mempool::load_mempool(&db.cauldron_w).await?; let electrum_clone = electrum.clone(); - let (cauldron_txs, oracle_txs, ido_txs) = tokio::task::spawn_blocking(move || { + let (cauldron_txs, oracle_txs, ido_txs, bcmr_txs) = tokio::task::spawn_blocking(move || { electrum_fetch_mempool(&electrum_clone.lock().unwrap()) }) .await??; @@ -218,6 +218,47 @@ pub async fn update_mempool( let txs_to_add = ttor_sorted_kahn(txs_to_add); db::ido::index_txs(network, &db.ido_w, &txs_to_add, None).await?; + // bcmr updates: index newly registered BCMRs without waiting for a confirmation + + // Drop mempool-indexed entries whose tx left the mempool: either it + // confirmed (index_blocks already re-stamped the entry with the real + // blockhash, so it no longer matches the sentinel) or it was evicted or + // replaced and must not linger as a stale auth head. + for txid in db::bcmr::get_unconfirmed_txids(&db.bcmr_w).await? { + if !bcmr_txs.contains(&txid) { + db::bcmr::delete_unconfirmed_tx(&db.bcmr_w, &txid).await?; + } + } + + // filter txs already indexed into the auth chain + let mut bcmr_to_add = Vec::new(); + for txid in bcmr_txs { + if !db::bcmr::has_indexed_tx(&db.bcmr_w, &txid).await? { + bcmr_to_add.push(txid); + } + } + + let bcmr_electrum = electrum.clone(); + let txs_to_add: Vec = tokio::task::spawn_blocking(move || { + bcmr_to_add + .into_iter() + .filter_map( + |txid| match electrum_get_tx(&bcmr_electrum.lock().unwrap(), &txid) { + Ok(tx) => Some(tx), + Err(e) => { + info!("Failed to get mempool tx {txid}: {e}"); + None + } + }, + ) + .collect() + }) + .await?; + + // an auth chain can have several unconfirmed txs in flight; index parents first + let txs_to_add = ttor_sorted_kahn(txs_to_add); + index_bcmr(&db.bcmr_w, &BlockHash::all_zeros(), txs_to_add).await?; + Ok(()) } diff --git a/src/main.rs b/src/main.rs index 29f76c5..952ea37 100644 --- a/src/main.rs +++ b/src/main.rs @@ -232,6 +232,7 @@ async fn start_program( db::oracle::clear_mempool(&db.oracle_w).await.unwrap(); db::moria::clear_mempool(&db.moria_w).await.unwrap(); + db::bcmr::clear_mempool(&db.bcmr_w).await.unwrap(); let indexing_in_progress_clone = indexing_in_progress.clone(); let ibd_state_clone = ibd_state.clone();