VMX-CORDINATOR-005

src/speedup.rs

@@ -33,8 +33,6 @@ pub trait SpeedupStore {
         txid: &Txid,
     ) -> Result<CoordinatedSpeedUpTransaction, BitcoinCoordinatorStoreError>;
 
-    fn can_speedup(&self) -> Result<bool, BitcoinCoordinatorStoreError>;
-
     fn is_funding_available(&self) -> Result<bool, BitcoinCoordinatorStoreError>;
 
     fn has_enough_unconfirmed_txs_for_cpfp(&self) -> Result<bool, BitcoinCoordinatorStoreError>;
@@ -88,9 +86,10 @@ impl SpeedupStore for BitcoinCoordinatorStore {
         // The broadcast block height is set to 0 and Finalized because funding should be confirmed on chain.
         let funding_to_speedup = CoordinatedSpeedUpTransaction::new(
             next_funding.txid,
-            next_funding.clone(),
-            next_funding,
-            None, // Funding is not an RBF replacement
+            None,                 // Funding transactions don't have an associated speedup tx
+            next_funding.clone(), // prev_funding
+            next_funding,         // next_funding
+            None,                 // Funding is not an RBF replacement
             0,
             TransactionState::Finalized,
             1.0,
@@ -281,19 +280,6 @@ impl SpeedupStore for BitcoinCoordinatorStore {
         Ok(active_speedups)
     }
 
-    /// Determines if a speedup (CPFP) transaction can be created and dispatched.
-    ///
-    /// Returns `true` if:
-    ///   - There is a funding transaction available to pay for the speedup.
-    ///   - There are enough available unconfirmed transaction slots to satisfy Bitcoin's mempool chain limit policy.
-    ///     (At least `MIN_UNCONFIRMED_TXS_FOR_CPFP` unconfirmed transactions are required: one for the CPFP itself and at least one unconfirmed output to spend.)
-    fn can_speedup(&self) -> Result<bool, BitcoinCoordinatorStoreError> {
-        let is_funding_available = self.is_funding_available()?;
-        let is_enough_unconfirmed_txs = self.has_enough_unconfirmed_txs_for_cpfp()?;
-
-        Ok(is_funding_available && is_enough_unconfirmed_txs)
-    }
-
     fn is_funding_available(&self) -> Result<bool, BitcoinCoordinatorStoreError> {
         let funding = self.get_funding()?;
         let is_funding_available = funding.is_some();

src/storage.rs

@@ -258,10 +258,19 @@ impl BitcoinCoordinatorStoreApi for BitcoinCoordinatorStore {
 
         // Validate state transitions
         let valid_transition = match (&tx.state, &new_state) {
-            // Valid transitions
             (TransactionState::ToDispatch, TransactionState::InMempool) => true,
             (TransactionState::ToDispatch, TransactionState::Failed) => true,
+
+            // From ToDispatch to Confirmed, Finalized is possible if the client crash and in a
+            // new tick detects that the transaction was already in the chain and is confirmed or finalized.
+            // Same happens with InMempool to Confirmed, Finalized.
+            (TransactionState::ToDispatch, TransactionState::Confirmed) => true,
+            (TransactionState::ToDispatch, TransactionState::Finalized) => true,
+
             (TransactionState::InMempool, TransactionState::Confirmed) => true,
+            (TransactionState::InMempool, TransactionState::ToDispatch) => true,
+            (TransactionState::InMempool, TransactionState::Finalized) => true,
+
             (TransactionState::Confirmed, TransactionState::Finalized) => true,
             // Allow transition from Confirmed to InMempool when transaction becomes orphan (reorg)
             (TransactionState::Confirmed, TransactionState::InMempool) => true,

src/types.rs

@@ -1,6 +1,7 @@
 use bitcoin::{Transaction, Txid};
 use bitvmx_bitcoin_rpc::types::BlockHeight;
 use bitvmx_transaction_monitor::types::{AckMonitorNews, MonitorNews};
+use bitvmx_transaction_monitor::TransactionBlockchainStatus;
 use protocol_builder::types::{output::SpeedupData, Utxo};
 use serde::{Deserialize, Serialize};
 
@@ -72,6 +73,9 @@ pub struct CoordinatedSpeedUpTransaction {
 
     pub tx_id: Txid,
 
+    // The speedup transaction itself (needed for dispatch, None for funding transactions)
+    pub tx: Option<Transaction>,
+
     // The previous funding utxo.
     pub prev_funding: Utxo,
 
@@ -124,6 +128,7 @@ impl RetryInfo {
 impl CoordinatedSpeedUpTransaction {
     pub fn new(
         tx_id: Txid,
+        tx: Option<Transaction>,
         prev_funding: Utxo,
         next_funding: Utxo,
         replaces_tx_id: Option<Txid>,
@@ -154,6 +159,7 @@ impl CoordinatedSpeedUpTransaction {
                 SpeedupType::CPFP
             },
             tx_id,
+            tx,
             prev_funding,
             next_funding,
             replaced_by_tx_id: None, // Initially, no transaction replaces this one
@@ -196,6 +202,19 @@ impl CoordinatedSpeedUpTransaction {
     }
 }
 
+/// Direct, enum-to-enum mapping between the indexer blockchain status and our internal TransactionState.
+impl From<TransactionBlockchainStatus> for TransactionState {
+    fn from(status: TransactionBlockchainStatus) -> Self {
+        match status {
+            TransactionBlockchainStatus::InMempool => TransactionState::InMempool,
+            TransactionBlockchainStatus::Confirmed => TransactionState::Confirmed,
+            TransactionBlockchainStatus::Finalized => TransactionState::Finalized,
+            TransactionBlockchainStatus::NotFound => TransactionState::ToDispatch,
+            TransactionBlockchainStatus::Orphan => TransactionState::InMempool,
+        }
+    }
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct TransactionFullInfo {
     pub tx: Transaction,

tests/batch_txs_test.rs

@@ -17,7 +17,10 @@ fn batch_txs_regtest_test() -> Result<(), anyhow::Error> {
     config_trace_aux();
 
     let mut blocks_mined = 102;
-    info!("Starting batch_txs_regtest_test with {} initial blocks mined", blocks_mined);
+    info!(
+        "Starting batch_txs_regtest_test with {} initial blocks mined",
+        blocks_mined
+    );
     let setup = create_test_setup(TestSetupConfig {
         blocks_mined,
         bitcoind_flags: None,
@@ -151,6 +154,12 @@ fn batch_txs_regtest_test() -> Result<(), anyhow::Error> {
 
     info!("Ticking coordinator after second block mined");
     coordinator.tick()?;
+    coordinator.tick()?;
+    setup
+        .bitcoin_client
+        .mine_blocks_to_address(1, &setup.funding_wallet)?;
+    coordinator.tick()?;
+    coordinator.tick()?;
 
     let news = coordinator.get_news()?;
     info!(

tests/reorg_rbf_test.rs

@@ -12,7 +12,9 @@ use protocol_builder::types::Utxo;
 use std::rc::Rc;
 use tracing::info;
 
-use crate::utils::{config_trace_aux, coordinate_tx, create_test_setup, TestSetupConfig};
+use crate::utils::{
+    config_trace_aux, coordinate_tx, create_test_setup, tick_until_coordinator_ready, TestSetupConfig,
+};
 mod utils;
 
 #[test]
@@ -109,7 +111,7 @@ fn replace_speedup_regtest_test() -> Result<(), anyhow::Error> {
 
     // Mine 3 blocks to confirm tx1 and its speedup transaction
     // Each block mined advances the blockchain, and each tick processes the new blocks
-    for _ in 0..3 {
+    for _ in 0..6 {
         info!("{} Mine and Tick", style("Test").green());
         setup
             .bitcoin_client
@@ -120,6 +122,7 @@ fn replace_speedup_regtest_test() -> Result<(), anyhow::Error> {
 
     // Verify that tx1 has been confirmed (1 confirmation)
     let news = coordinator.get_news()?;
+
     assert_eq!(
         news.monitor_news.len(),
         1,
@@ -224,9 +227,7 @@ fn replace_speedup_regtest_test() -> Result<(), anyhow::Error> {
         .unwrap();
 
     // Wait for coordinator to be ready (indexer synced with blockchain)
-    while !coordinator.is_ready()? {
-        coordinator.tick()?;
-    }
+    tick_until_coordinator_ready(&coordinator)?;
 
     // After mining a new block, tx1 should be confirmed again (re-mined in the new chain)
     let news = coordinator.get_news()?;

tests/reorg_test.rs

@@ -82,7 +82,7 @@ fn replace_speedup_regtest_test() -> Result<(), anyhow::Error> {
 
     coordinator.tick()?;
 
-    for _ in 0..4 {
+    for _ in 0..8 {
         info!("{} Mine and Tick", style("Test").green());
         // Mine a block to confirm tx1 and its speedup transaction
         setup

tests/replace_speedup_regtest_test.rs

@@ -7,7 +7,9 @@ use protocol_builder::types::Utxo;
 use std::rc::Rc;
 use tracing::info;
 
-use crate::utils::{config_trace_aux, coordinate_tx, create_test_setup, TestSetupConfig};
+use crate::utils::{
+    config_trace_aux, coordinate_tx, create_test_setup, tick_until_coordinator_ready, TestSetupConfig,
+};
 mod utils;
 
 // Almost every transaction sent in the protocol uses a CPFP (Child Pays For Parent) transaction for broadcasting.
@@ -66,12 +68,8 @@ fn replace_speedup_regtest_test() -> Result<(), anyhow::Error> {
         None,
     )?);
 
-    // Since we've already mined 102 blocks, we need to advance the coordinator by 102 ticks
-    // so the indexer can catch up with the current blockchain height.
     // Tick coordinator until it is ready (indexer is caught up with the current blockchain height)
-    while !coordinator.is_ready()? {
-        coordinator.tick()?;
-    }
+    tick_until_coordinator_ready(&coordinator)?;
 
     // Add funding for speed up transaction
     coordinator.add_funding(Utxo::new(

tests/speedup_chain_recompute_fee_test.rs

@@ -129,10 +129,12 @@ fn speedup_chain_recompute_fee_test() -> Result<(), anyhow::Error> {
         coordinator.tick()?;
     }
 
-    setup
-        .bitcoin_client
-        .mine_blocks_to_address(1, &setup.funding_wallet)?;
-    coordinator.tick()?;
+    for _ in 0..4 {
+        setup
+            .bitcoin_client
+            .mine_blocks_to_address(1, &setup.funding_wallet)?;
+        coordinator.tick()?;
+    }
 
     let news = coordinator.get_news()?;
     assert_eq!(news.monitor_news.len(), 1);

tests/speedup_prevalidation_test.rs

@@ -0,0 +1,160 @@
+use bitcoin::Amount;
+use bitcoin_coordinator::{
+    coordinator::{BitcoinCoordinator, BitcoinCoordinatorApi},
+    storage::{BitcoinCoordinatorStore, BitcoinCoordinatorStoreApi},
+    types::TransactionState,
+};
+use bitvmx_bitcoin_rpc::bitcoin_client::BitcoinClientApi;
+use protocol_builder::types::Utxo;
+use std::rc::Rc;
+use tracing::info;
+
+mod utils;
+use crate::utils::{
+    config_trace_aux, coordinate_tx, create_test_setup, tick_until_coordinator_ready,
+    TestSetupConfig,
+};
+
+/// Indices (0-based) of the 4 intentionally invalid transactions among 40 (not consecutive).
+const INVALID_TX_INDICES: [usize; 4] = [3, 11, 19, 27];
+
+/// Integration test for the speedup pre‑validation + batching path:
+/// - Creates 40 coordinated transactions; 4 of them use fee 0 so `testmempoolaccept` rejects them.
+/// - `filter_txs_allowed_by_mempool` must mark those 4 as `Failed` and only keep the 36 valid ones.
+/// - The 36 valid txs are dispatched in 2 CPFP batches and later reported as monitor news.
+#[test]
+fn speedup_prevalidation_40_txs_4_invalid_two_batches() -> Result<(), anyhow::Error> {
+    config_trace_aux();
+
+    let mut blocks_mined = 102;
+    let setup = create_test_setup(TestSetupConfig {
+        blocks_mined,
+        bitcoind_flags: None,
+    })?;
+
+    let amount = Amount::from_sat(23450000);
+
+    let _ = setup
+        .bitcoin_client
+        .fund_address(&setup.funding_wallet, amount)?;
+    blocks_mined += 1;
+
+    let (funding_speedup, funding_speedup_vout) = setup
+        .bitcoin_client
+        .fund_address(&setup.funding_wallet, amount)?;
+    blocks_mined += 1;
+
+    let coordinator = Rc::new(BitcoinCoordinator::new_with_paths(
+        &setup.config_bitcoin_client,
+        setup.storage.clone(),
+        setup.key_manager.clone(),
+        None,
+    )?);
+
+    for _ in 0..blocks_mined {
+        coordinator.tick()?;
+    }
+
+    coordinator.add_funding(Utxo::new(
+        funding_speedup.compute_txid(),
+        funding_speedup_vout,
+        amount.to_sat(),
+        &setup.public_key,
+    ))?;
+
+    // Build 40 coordinated txs; at indices in INVALID_TX_INDICES we force fee = 0 so
+    // `testmempoolaccept` rejects them during pre‑validation. The rest use the default fee.
+    let mut invalid_tx_ids = Vec::new();
+    info!(
+        "Submitting 40 transactions (4 invalid at indices {:?})",
+        INVALID_TX_INDICES
+    );
+    for i in 0..40 {
+        let is_invalid = INVALID_TX_INDICES.contains(&i);
+        let fee = if is_invalid { Some(0) } else { None };
+        let tx = coordinate_tx(
+            coordinator.clone(),
+            amount,
+            setup.network,
+            setup.key_manager.clone(),
+            setup.bitcoin_client.clone(),
+            fee,
+        )?;
+        if is_invalid {
+            invalid_tx_ids.push(tx.compute_txid());
+        }
+    }
+
+    tick_until_coordinator_ready(&coordinator)?;
+
+    // First batch:
+    // Mine + tick twice so the coordinator first processes parent txs and then their CPFP speedups,
+    // and the monitor accumulates news for the first batch of valid transactions.
+    setup
+        .bitcoin_client
+        .mine_blocks_to_address(1, &setup.funding_wallet)?;
+
+    coordinator.tick()?;
+
+    setup
+        .bitcoin_client
+        .mine_blocks_to_address(1, &setup.funding_wallet)?;
+
+    coordinator.tick()?;
+
+    let news = coordinator.get_news()?;
+    let first_batch_count = news.monitor_news.len();
+    info!(
+        "After first block: monitor_news.len() = {}",
+        first_batch_count
+    );
+    assert!(
+        first_batch_count == 24,
+        "expected first batch 24 monitor notifications, got {}",
+        first_batch_count
+    );
+
+    tick_until_coordinator_ready(&coordinator)?;
+
+    // Second batch:
+    // Again, mine + tick twice so the remaining valid transactions are confirmed and reported.
+    setup
+        .bitcoin_client
+        .mine_blocks_to_address(1, &setup.funding_wallet)?;
+
+    coordinator.tick()?;
+
+    setup
+        .bitcoin_client
+        .mine_blocks_to_address(1, &setup.funding_wallet)?;
+
+    coordinator.tick()?;
+
+    let news = coordinator.get_news()?;
+    info!(
+        "After second block: monitor_news.len() = {}",
+        news.monitor_news.len()
+    );
+    assert_eq!(
+        news.monitor_news.len(),
+        36,
+        "expected 36 monitor notifications (only valid txs), got {}",
+        news.monitor_news.len()
+    );
+
+    // 4 invalid txs must be Failed in store (not active).
+    let store = BitcoinCoordinatorStore::new(setup.storage.clone(), 10)?;
+    for tx_id in &invalid_tx_ids {
+        let coordinated = store.get_tx(tx_id)?;
+        assert_eq!(
+            coordinated.state,
+            TransactionState::Failed,
+            "invalid tx {} should be Failed",
+            tx_id
+        );
+    }
+
+    setup.bitcoind.stop()?;
+
+    Ok(())
+}