ido: index unconfirmed txs from the mempool

- electrum mempool fetch gains an ido filter: state machine spends
  (IDO_SIGNATURE in scriptsig) union preinit announcements
  (IDO_PREINIT_ANNOUNCEMENT_SIGNATURE in scriptpubkey)
- split tx scanning out of index_block into index_txs and add
  index_mempool, which indexes with a -1 sentinel height;
  txchain_trim_tracker leaves negative heights alone and the real
  height replaces the sentinel once the tx confirms in a block
- block indexing now skips re-creating an ido already seen in the
  mempool, only updating its tracker height
- mempool txs are kahn-sorted so txchain parents index before children

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/db/ido/mod.rs

@@ -82,7 +82,9 @@ const CHIPNET_PLATFORM_FEE_NFTH: LazyLock<Vec<u8>> = LazyLock::new(|| hex::decod
 // TODO:: set a mock nfthash for mainnet
 const MAINNET_PLATFORM_FEE_NFTH: LazyLock<Vec<u8>> = LazyLock::new(|| hex::decode("").unwrap());
 
-const IDO_SIGNATURE: &[u8] = &[
+// (OP_PUSH18 "CauldronIdo-2026Q2" OP_DROP), found at the start of every ido
+// state machine redeem script.
+pub const IDO_SIGNATURE: &[u8] = &[
     0x12, 0x43, 0x61, 0x75, 0x6c, 0x64, 0x72, 0x6f, 0x6e, 0x49,
     0x64, 0x6f, 0x2d, 0x32, 0x30, 0x32, 0x36, 0x51, 0x32, 0x75,
 ];
@@ -910,7 +912,9 @@ async fn txchain_trim_tracker(
     Ok(())
 }
 
-const IDO_PREINIT_ANNOUNCEMENT_SIGNATURE: &[u8; 15] = &[
+// Prefix of the announcement OP_RETURN carried in the last output of a preinit
+// broadcast tx.
+pub const IDO_PREINIT_ANNOUNCEMENT_SIGNATURE: &[u8; 15] = &[
     0x6a, 0x4c, 0xbb, // OP_RETURN OP_PUSHDATA1 (187)
     0x43, 0x61, 0x75, 0x6c, 0x64, 0x72, 0x6f, 0x6e, 0x49, 0x64, 0x6f, 0x30, // CauldronIdo0
 ];
@@ -2213,17 +2217,23 @@ fn is_ido_sig_in_tx_inputs(tx: &Transaction) -> bool {
     return false;
 }
 
-pub async fn index_block(
+async fn index_txs(
     network: Option<Network>,
     pool: &SqlitePool,
     sorted_txs: &[Transaction],
-    _blockhash: &BlockHash,
-    _mtp: i64,
     block_height: i64,
 ) -> Result<()> {
     for tx in sorted_txs {
         // detect a new ido
         if is_preinit_broadcast(tx) {
+            if lookup_internal_id_by_preinit_txid(pool, &tx.txid().to_blob()).await?.is_some() {
+                // already created (eg. indexed from the mempool), only update block height
+                let mut dbtx = pool.begin().await?;
+                update_ido_txchain_tracker_block_height(&mut dbtx, tx, block_height)
+                    .await?;
+                dbtx.commit().await?;
+                continue;
+            }
             match on_create_ido(network, pool, tx, block_height)
                 .await {
                     Err(err) => info!("failed to detect an ido, or an invalid ido detected, txid: {}, {}", hex::encode(tx.txid().to_blob()), err),
@@ -2260,11 +2270,46 @@ pub async fn index_block(
             }
         }
     }
+    Ok(())
+}
+
+pub async fn index_block(
+    network: Option<Network>,
+    pool: &SqlitePool,
+    sorted_txs: &[Transaction],
+    _blockhash: &BlockHash,
+    _mtp: i64,
+    block_height: i64,
+) -> Result<()> {
+    index_txs(network, pool, sorted_txs, block_height)
+        .await?;
     txchain_trim_tracker(pool, block_height)
         .await?;
     Ok(())
 }
 
+// Sentinel block height for txs indexed from the mempool. txchain_trim_tracker
+// never trims tracker entries with a negative height; the height is replaced
+// with the real one once the tx confirms in a block.
+const MEMPOOL_BLOCK_HEIGHT: i64 = -1;
+
+pub async fn index_mempool(
+    network: Option<Network>,
+    pool: &SqlitePool,
+    sorted_txs: &[Transaction],
+) -> Result<()> {
+    index_txs(network, pool, sorted_txs, MEMPOOL_BLOCK_HEIGHT)
+        .await
+}
+
+/// Whether the tx is already part of an indexed ido txchain.
+pub async fn has_txchain_tx(
+    pool: &SqlitePool,
+    txid: &Txid,
+) -> Result<bool> {
+    Ok(get_txchain_item_by_txid(pool, txid).await?.is_some())
+}
+
 // ─── Public RPC types ────────────────────────────────────────────────────────
 
 #[derive(Serialize, Clone)]
@@ -2778,6 +2823,7 @@ mod querytests {
             idoCategory: vec![0xBB; 32],
             nextTxSetterFlag: BigInt::from(0i64),
             oTokenGenerated: BigInt::from(1i64),
+            initiatorCreated: false,
         }
     }
 

src/electrum.rs

@@ -18,6 +18,8 @@ use riftenlabs_defi::{
 };
 use serde_json::{json, Value};
 
+use crate::db::ido::{IDO_PREINIT_ANNOUNCEMENT_SIGNATURE, IDO_SIGNATURE};
+
 /// Fetch blockchain tip from electrum server
 pub fn electrum_get_tip(client: &Client) -> Result<(BlockHeader, u64)> {
     let tip: Value =
@@ -39,7 +41,7 @@ pub fn electrum_get_tip(client: &Client) -> Result<(BlockHeader, u64)> {
 }
 
 /// Fetch defi (cauldron + tokentoken) and oracle mempool transactions
-pub fn electrum_fetch_mempool(client: &Client) -> Result<(HashSet<Txid>, HashSet<Txid>)> {
+pub fn electrum_fetch_mempool(client: &Client) -> Result<(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)
@@ -72,6 +74,12 @@ pub fn electrum_fetch_mempool(client: &Client) -> Result<(HashSet<Txid>, HashSet
         "scriptsig": hex::encode(DELPHI_V2_REDEEM_SCRIPT_BODY),
     });
 
+    let ido_filter = json!({
+        "scriptsig": hex::encode(IDO_SIGNATURE), // ido state machine spends
+        "scriptpubkey": hex::encode(IDO_PREINIT_ANNOUNCEMENT_SIGNATURE), // preinit announcements (potentially)
+        "operation": "union"
+    });
+
     let fetch_txs = |filter: Value| -> Result<HashSet<Txid>> {
         let response = client.raw_call("mempool.get", [Param::Value(filter)])?;
 
@@ -104,8 +112,9 @@ pub fn electrum_fetch_mempool(client: &Client) -> Result<(HashSet<Txid>, HashSet
     defi_txs.extend(fetch_txs(tokentoken_filter)?);
     let mut oracle_txs = fetch_txs(oracle_v1_filter)?;
     oracle_txs.extend(fetch_txs(oracle_v2_filter)?);
+    let ido_txs = fetch_txs(ido_filter)?;
 
-    Ok((defi_txs, oracle_txs))
+    Ok((defi_txs, oracle_txs, ido_txs))
 }
 
 /// Fetch blockchain tip from electrum server

src/index.rs

@@ -67,12 +67,20 @@ fn moria_token_ids(network: Option<Network>) -> MoriaTokenIds {
     }
 }
 
-pub async fn update_mempool(db: &DB, electrum: Arc<Mutex<Client>>) -> Result<()> {
+pub async fn update_mempool(
+    db: &DB,
+    electrum: Arc<Mutex<Client>>,
+    network: Option<Network>,
+) -> Result<()> {
     let our_mempool_txs: HashSet<Txid> =
         db::cauldron::mempool::load_mempool(&db.cauldron_w).await?;
 
     let electrum_clone = electrum.clone();
+<<<<<<< HEAD
     let (defi_txs, oracle_txs) = tokio::task::spawn_blocking(move || {
+=======
+    let (cauldron_txs, oracle_txs, ido_txs) = tokio::task::spawn_blocking(move || {
+>>>>>>> 7fb15f6 (ido: index unconfirmed txs from the mempool)
         electrum_fetch_mempool(&electrum_clone.lock().unwrap())
     })
     .await??;
@@ -182,6 +190,37 @@ pub async fn update_mempool(db: &DB, electrum: Arc<Mutex<Client>>) -> Result<()>
 
     index_oracle(&db.oracle_w, &txs_to_add, &BlockHash::all_zeros()).await?;
 
+    // ido updates
+    let ido_electrum = electrum.clone();
+
+    // filter txs already in an ido txchain
+    let mut ido_to_add = Vec::new();
+    for txid in ido_txs {
+        if !db::ido::has_txchain_tx(&db.ido_w, &txid).await? {
+            ido_to_add.push(txid);
+        }
+    }
+
+    let txs_to_add: Vec<Transaction> = tokio::task::spawn_blocking(move || {
+        ido_to_add
+            .into_iter()
+            .filter_map(
+                |txid| match electrum_get_tx(&ido_electrum.lock().unwrap(), &txid) {
+                    Ok(tx) => Some(tx),
+                    Err(e) => {
+                        info!("Failed to get mempool tx {txid}: {e}");
+                        None
+                    }
+                },
+            )
+            .collect()
+    })
+    .await?;
+
+    // an ido txchain can have several unconfirmed txs in flight; index parents first
+    let txs_to_add = ttor_sorted_kahn(txs_to_add);
+    db::ido::index_mempool(network, &db.ido_w, &txs_to_add).await?;
+
     Ok(())
 }
 

src/main.rs

@@ -338,7 +338,7 @@ async fn start_program(
 
             // Avoid overlapping writer while indexer is on
             if !indexing_in_progress_clone.load(Ordering::Relaxed) {
-                if let Err(e) = update_mempool(&db, client.clone()).await {
+                if let Err(e) = update_mempool(&db, client.clone(), Some(network)).await {
                     error!("Failed to update mempool: {e}");
                 }
             }