Compare commits
12 commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0eb86131f4 | ||
|
|
a73dff52df | ||
|
|
da26864cba | ||
|
|
c1952142de | ||
|
|
4677cc955e | ||
|
|
c8450e210b | ||
|
|
ca7f829591 | ||
|
|
c3f3c269e2 | ||
|
|
2bcf35172c | ||
|
|
a5f4ce40d4 | ||
|
|
1ddab462aa | ||
|
|
d363c2c930 |
6 changed files with 1135 additions and 380 deletions
|
|
@ -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();
|
||||
|
|
|
|||
1255
src/db/ido/mod.rs
1255
src/db/ido/mod.rs
File diff suppressed because one or more lines are too long
|
|
@ -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) =
|
||||
|
|
@ -142,6 +144,9 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul
|
|||
create_db_pool(&db_path(db_dir, "ido.db"), read_slots.ido).await;
|
||||
if !db_exists {
|
||||
ido_prepare_tables(&ido_db_write).await;
|
||||
crate::db::ido::set_db_version(&ido_db_write).await?;
|
||||
} else {
|
||||
crate::db::ido::check_db_version(&ido_db_read).await?;
|
||||
}
|
||||
|
||||
Ok(DB {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
43
src/index.rs
43
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(())
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue