Merge branch 'crc20-fix' into 'master'

Fix cross-database query bug in CRC20 BCMR filtering

See merge request riftenlabs/riftenlabs-indexer!61

src/crc20/crc20fetcher.rs

@@ -9,8 +9,10 @@ use std::{
     time::Duration,
 };
 
+use crate::db::bcmr::filter_tokens_with_bcmr;
 use crate::db::crc20::{
-    bump_failed_attempts, get_not_indexed_tokens, update_to_crc20, update_to_not_crc20,
+    bump_failed_attempts, get_not_indexed_tokens, mark_as_has_bcmr, update_to_crc20,
+    update_to_not_crc20,
 };
 use crate::db::DBPool;
 use anyhow::*;
@@ -18,6 +20,7 @@ use electrum_client_netagnostic::{Client, ElectrumApi, Param};
 use log::{info, warn};
 use rand::thread_rng;
 use serde_json::Value;
+use std::collections::HashSet;
 
 use rand::seq::SliceRandom;
 use std::result::Result::Ok;
@@ -67,7 +70,8 @@ impl CRC20Fetcher {
 
     pub fn start(
         &mut self,
-        db: DBPool,
+        crc20_db: DBPool,
+        bcmr_db: DBPool,
         electrum: Arc<Mutex<Client>>,
         indexing_flag: Arc<AtomicBool>,
     ) -> Result<()> {
@@ -88,8 +92,9 @@ impl CRC20Fetcher {
                         continue;
                     }
 
-                    let mut queue = {
-                        let db = match db.get() {
+                    // Get candidate tokens from crc20 database
+                    let candidates: Vec<(Vec<u8>, String)> = {
+                        let db = match crc20_db.get() {
                             Ok(db) => db,
                             Err(e) => {
                                 warn!("Failed to get crc20 db connection {e}");
@@ -107,6 +112,61 @@ impl CRC20Fetcher {
                         }
                     };
 
+                    if candidates.is_empty() {
+                        thread::sleep(LOOP_SLEEP_TIME);
+                        continue;
+                    }
+
+                    // Check which candidates already have BCMR data
+                    let token_blobs: Vec<Vec<u8>> =
+                        candidates.iter().map(|(blob, _)| blob.clone()).collect();
+                    let tokens_with_bcmr: HashSet<Vec<u8>> = {
+                        let db = match bcmr_db.get() {
+                            Ok(db) => db,
+                            Err(e) => {
+                                warn!("Failed to get bcmr db connection {e}");
+                                thread::sleep(LOOP_SLEEP_TIME);
+                                continue;
+                            }
+                        };
+                        match filter_tokens_with_bcmr(&db, &token_blobs) {
+                            Ok(tokens) => tokens.into_iter().collect(),
+                            Err(e) => {
+                                warn!("Failed to filter tokens with bcmr {e}");
+                                thread::sleep(LOOP_SLEEP_TIME);
+                                continue;
+                            }
+                        }
+                    };
+
+                    // Mark tokens with BCMR in crc20 database so they're skipped in future
+                    if !tokens_with_bcmr.is_empty() {
+                        let bcmr_blobs: Vec<Vec<u8>> = tokens_with_bcmr.iter().cloned().collect();
+                        let db = match crc20_db.get() {
+                            Ok(db) => db,
+                            Err(e) => {
+                                warn!("Failed to get crc20 db connection for marking bcmr {e}");
+                                thread::sleep(LOOP_SLEEP_TIME);
+                                continue;
+                            }
+                        };
+                        if let Err(e) = mark_as_has_bcmr(&db, &bcmr_blobs) {
+                            warn!("Failed to mark tokens as having bcmr {e}");
+                        } else {
+                            info!(
+                                "crc20: marked {} tokens as having BCMR data",
+                                bcmr_blobs.len()
+                            );
+                        }
+                    }
+
+                    // Filter out tokens that have BCMR data
+                    let mut queue: Vec<String> = candidates
+                        .into_iter()
+                        .filter(|(blob, _)| !tokens_with_bcmr.contains(blob))
+                        .map(|(_, hex)| hex)
+                        .collect();
+
                     // in case electrum has issues with a token; with rng we'll eventually get all others
                     let mut rng = thread_rng();
                     queue.shuffle(&mut rng);
@@ -138,7 +198,7 @@ impl CRC20Fetcher {
                             }
                         };
 
-                        let db = match db.get() {
+                        let db = match crc20_db.get() {
                             Ok(db) => db,
                             Err(e) => {
                                 warn!("Failed to get crc20 db connection {e}");

src/db/bcmr/mod.rs

@@ -483,6 +483,40 @@ pub fn get_well_known_bcmr(conn: &Connection, token_hex: &str) -> Result<Vec<Par
     Ok(entries)
 }
 
+/// Given a list of token blobs, returns which ones have BCMR data in bcmr_data table.
+/// Uses batched IN queries to handle large lists efficiently.
+pub fn filter_tokens_with_bcmr(conn: &Connection, token_blobs: &[Vec<u8>]) -> Result<Vec<Vec<u8>>> {
+    if token_blobs.is_empty() {
+        return Ok(Vec::new());
+    }
+
+    let mut result = Vec::new();
+
+    // Batch queries to stay within SQLite parameter limits
+    const BATCH_SIZE: usize = 500;
+
+    for chunk in token_blobs.chunks(BATCH_SIZE) {
+        let placeholders: Vec<&str> = chunk.iter().map(|_| "?").collect();
+        let sql = format!(
+            "SELECT DISTINCT token_id FROM bcmr_data WHERE token_id IN ({})",
+            placeholders.join(", ")
+        );
+
+        let params: Vec<&dyn rusqlite::ToSql> =
+            chunk.iter().map(|b| b as &dyn rusqlite::ToSql).collect();
+
+        let mut stmt = conn.prepare(&sql)?;
+        let mut rows = stmt.query(params.as_slice())?;
+
+        while let Some(row) = rows.next()? {
+            let blob: Vec<u8> = row.get(0)?;
+            result.push(blob);
+        }
+    }
+
+    Ok(result)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/db/crc20/mod.rs

@@ -13,6 +13,7 @@ use crate::db::blob::{blob_to_display_hex, display_hex_to_blob, ToBlob};
 const STATE_NOT_INDEXED: i32 = -1;
 const STATE_NOT_CRC20: i32 = 0;
 const STATE_IS_CRC20: i32 = 1;
+const STATE_SKIP_HAS_BCMR: i32 = 2;
 
 const MAX_FAILED_ATTEMPTS: i32 = 20;
 
@@ -56,29 +57,48 @@ pub fn insert_crc20_candidate(conn: &Connection, token_id: &TokenID) -> Result<(
     Ok(())
 }
 
-pub fn get_not_indexed_tokens(conn: &Connection) -> Result<Vec<String>> {
-    // Skip tokens that already have valid BCMR metadata (entry in bcmr_data table).
-    // In practice there is little overlap between BCMR and CRC20 tokens, but if
-    // BCMR metadata exists we prefer it and skip the CRC20 fetch.
+/// Returns token blobs and their hex representations for tokens that need CRC20 indexing.
+/// Tokens marked as STATE_SKIP_HAS_BCMR are excluded.
+pub fn get_not_indexed_tokens(conn: &Connection) -> Result<Vec<(Vec<u8>, String)>> {
     let mut stmt = conn.prepare(
-        "SELECT c.token_id
-         FROM crc20_candidates c
-         WHERE c.is_crc20 = ?1
-           AND c.failed_attempts <= ?2
-           AND NOT EXISTS (
-             SELECT 1 FROM bcmr_data b WHERE b.token_id = c.token_id
-           )",
+        "SELECT token_id
+         FROM crc20_candidates
+         WHERE is_crc20 = ?1
+           AND failed_attempts <= ?2",
     )?;
     let mut rows = stmt.query(params![STATE_NOT_INDEXED, MAX_FAILED_ATTEMPTS])?;
 
-    let mut token_ids = Vec::new();
+    let mut tokens = Vec::new();
     while let Some(row) = rows.next()? {
         let blob: Vec<u8> = row.get(0)?;
         let token_hex = blob_to_display_hex::<TokenID>(&blob)?;
-        token_ids.push(token_hex);
+        tokens.push((blob, token_hex));
     }
 
-    Ok(token_ids)
+    Ok(tokens)
+}
+
+/// Mark tokens as having BCMR data, so they are skipped in future CRC20 fetching.
+pub fn mark_as_has_bcmr(conn: &Connection, token_blobs: &[Vec<u8>]) -> Result<()> {
+    if token_blobs.is_empty() {
+        return Ok(());
+    }
+
+    let placeholders: Vec<&str> = token_blobs.iter().map(|_| "?").collect();
+    let sql = format!(
+        "UPDATE crc20_candidates SET is_crc20 = {} WHERE token_id IN ({})",
+        STATE_SKIP_HAS_BCMR,
+        placeholders.join(", ")
+    );
+
+    let params: Vec<&dyn rusqlite::ToSql> = token_blobs
+        .iter()
+        .map(|b| b as &dyn rusqlite::ToSql)
+        .collect();
+
+    conn.execute(&sql, params.as_slice())?;
+
+    Ok(())
 }
 
 pub fn update_to_not_crc20(conn: &Connection, token_hex: &str) -> Result<()> {
@@ -165,21 +185,18 @@ mod tests {
     use bitcoin_hashes::hex::FromHex;
 
     #[test]
-    fn test_get_not_indexed_tokens_excludes_bcmr_tokens() {
-        // Test that tokens with BCMR metadata are excluded from CRC20 fetching
+    fn test_get_not_indexed_tokens_excludes_bcmr_marked_tokens() {
+        // Test that tokens marked as having BCMR are excluded from CRC20 fetching
         let conn = Connection::open_in_memory().unwrap();
 
-        // Set up crc20 tables
+        // Set up crc20 tables only (no bcmr tables needed)
         prepare_tables(&conn);
 
-        // Set up bcmr tables (includes bcmr_data)
-        crate::db::bcmr::prepare_tables(&conn);
-
-        // Token A: has no BCMR data (should be returned)
+        // Token A: not marked as having BCMR (should be returned)
         let token_a =
             TokenID::from_hex("0101010101010101010101010101010101010101010101010101010101010101")
                 .unwrap();
-        // Token B: has BCMR data (should be excluded)
+        // Token B: will be marked as having BCMR (should be excluded)
         let token_b =
             TokenID::from_hex("0202020202020202020202020202020202020202020202020202020202020202")
                 .unwrap();
@@ -188,23 +205,26 @@ mod tests {
         insert_crc20_candidate(&conn, &token_a).unwrap();
         insert_crc20_candidate(&conn, &token_b).unwrap();
 
-        // Insert BCMR data for token B only
-        let utxo_blob: Vec<u8> = vec![0xaa; 32];
-        conn.execute(
-            "INSERT INTO bcmr_data (token_id, utxo, symbol, decimals, name, description, icon, web, expected_hash, actual_hash)
-             VALUES (?1, ?2, 'TST', 8, 'Test Token', 'A test token', '', '', '', '')",
-            params![token_b.to_blob(), utxo_blob],
-        )
-        .unwrap();
+        // Mark token B as having BCMR data
+        mark_as_has_bcmr(&conn, &[token_b.to_blob()]).unwrap();
 
         // Get not indexed tokens - should only return token A
         let result = get_not_indexed_tokens(&conn).unwrap();
 
         assert_eq!(result.len(), 1, "should only return 1 token");
         assert_eq!(
-            result[0].to_lowercase(),
+            result[0].1.to_lowercase(),
             token_a.to_hex().to_lowercase(),
-            "should return token_a (without BCMR data)"
+            "should return token_a (not marked as having BCMR)"
         );
     }
+
+    #[test]
+    fn test_mark_as_has_bcmr_empty_list() {
+        // Test that marking empty list doesn't error
+        let conn = Connection::open_in_memory().unwrap();
+        prepare_tables(&conn);
+
+        mark_as_has_bcmr(&conn, &[]).unwrap();
+    }
 }

src/main.rs

@@ -192,6 +192,7 @@ fn start_program(
     let mut crc20fetcher = CRC20Fetcher::new();
     crc20fetcher.start(
         db.crc20_w.clone(),
+        db.bcmr_r.clone(),
         client.clone(),
         indexing_in_progress.clone(),
     )?;