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 <noreply@anthropic.com>

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<bool> {
+    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<Vec<Txid>> {
+    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<u8> = 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<usize> {
+    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<bool> {
     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<Vec<u8>>,
 ) -> 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::<TokenID>()
+            .unwrap();
+        let txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+            .parse::<Txid>()
+            .unwrap();
+        let utxo = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+            .parse::<OutPointHash>()
+            .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::<BlockHash>()
+            .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<u8>,) = 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::<TokenID>()
+            .unwrap();
+        let txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+            .parse::<Txid>()
+            .unwrap();
+        let utxo = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+            .parse::<OutPointHash>()
+            .unwrap();
+        let real_blockhash = "1111111111111111111111111111111111111111111111111111111111111111"
+            .parse::<BlockHash>()
+            .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();

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) =

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<Txid>, HashSet<Txid>, HashSet<Txid>)> {
+) -> Result<(HashSet<Txid>, HashSet<Txid>, HashSet<Txid>, HashSet<Txid>)> {
     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<HashSet<Txid>> {
         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

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<Transaction> = 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(())
 }
 

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();