diff --git a/app/rpc_server.rs b/app/rpc_server.rs index faff067e..1ac22724 100644 --- a/app/rpc_server.rs +++ b/app/rpc_server.rs @@ -254,6 +254,22 @@ impl rpc_api::node::RpcServer Ok(peers) } + async fn list_mempool(&self) -> RpcResult> { + let txs = self.app.node.get_all_transactions().map_err(custom_err)?; + let res = txs + .into_iter() + .map(|authorized| { + let tx = authorized.transaction; + rpc_api::node::MempoolTx { + txid: tx.txid(), + size: tx.canonical_size(), + tx, + } + }) + .collect(); + Ok(res) + } + async fn list_utxos(&self) -> RpcResult> { let utxos = self.app.node.get_all_utxos().map_err(custom_err)?; let res = utxos diff --git a/cli/lib.rs b/cli/lib.rs index d324a872..464d1102 100644 --- a/cli/lib.rs +++ b/cli/lib.rs @@ -113,6 +113,8 @@ pub enum Command { }, /// Get the height of the latest failed withdrawal bundle LatestFailedWithdrawalBundleHeight, + /// List the transactions the mempool holds + ListMempool, /// List peers ListPeers, /// List all UTXOs @@ -298,6 +300,10 @@ where rpc_client.latest_failed_withdrawal_bundle_height().await?; serde_json::to_string_pretty(&height)? } + Command::ListMempool => { + let txs = rpc_client.list_mempool().await?; + serde_json::to_string_pretty(&txs)? + } Command::ListPeers => { let peers = rpc_client.list_peers().await?; serde_json::to_string_pretty(&peers)? diff --git a/integration_tests/integration_test.rs b/integration_tests/integration_test.rs index 5ff99bab..939d6f62 100644 --- a/integration_tests/integration_test.rs +++ b/integration_tests/integration_test.rs @@ -14,6 +14,7 @@ use thunder_app_rpc_api::node::RpcClient as _; use crate::{ block_template::block_template_trial, ibd::ibd_trial, + list_mempool::list_mempool_trial, setup::{Init, PostSetup}, unknown_withdrawal::unknown_withdrawal_trial, util::BinPaths, @@ -178,6 +179,11 @@ pub fn tests( file_registry.clone(), failure_collector.clone(), ), + list_mempool_trial( + bin_paths.clone(), + file_registry.clone(), + failure_collector.clone(), + ), unknown_withdrawal_trial(bin_paths, file_registry, failure_collector), ] } diff --git a/integration_tests/list_mempool.rs b/integration_tests/list_mempool.rs new file mode 100644 index 00000000..8846c154 --- /dev/null +++ b/integration_tests/list_mempool.rs @@ -0,0 +1,149 @@ +//! Test that a node lists the transactions its mempool holds + +use bip300301_enforcer_integration_tests::{ + integration_test::{ + activate_sidechain, deposit, fund_enforcer, propose_sidechain, + }, + setup::{ + Mode, Network, PostSetup as EnforcerPostSetup, + PreSetup as EnforcerPreSetup, SetupOpts as EnforcerSetupOpts, + Sidechain as _, + }, + util::{ + AbortOnDrop, AsyncTrial, BinPaths as EnforcerBinPaths, + TestFailureCollector, TestFileRegistry, + }, +}; +use bitcoin::Amount; +use futures::{ + FutureExt as _, StreamExt as _, channel::mpsc, future::BoxFuture, +}; +use thunder_app_rpc_api::{node::RpcClient as _, wallet::RpcClient as _}; +use tokio::time::sleep; +use tracing::Instrument as _; + +use crate::{ + setup::{Init, PostSetup}, + util::BinPaths, +}; + +const DEPOSIT_AMOUNT: Amount = Amount::from_sat(21_000_000); +const DEPOSIT_FEE: Amount = Amount::from_sat(1_000_000); +const TRANSFER_AMOUNT: u64 = 1_000_000; +const TRANSFER_FEE: u64 = 1_000; + +/// Initial setup for the test +async fn setup( + enforcer_bin_paths: &EnforcerBinPaths, + res_tx: mpsc::UnboundedSender>, +) -> anyhow::Result { + let enforcer_pre_setup = + EnforcerPreSetup::new(enforcer_bin_paths, Network::Regtest)?; + let mut enforcer_post_setup = { + let setup_opts: EnforcerSetupOpts = Default::default(); + enforcer_pre_setup + .setup(Mode::Mempool, setup_opts, res_tx.clone()) + .await? + }; + let () = propose_sidechain::(&mut enforcer_post_setup).await?; + let () = activate_sidechain::(&mut enforcer_post_setup).await?; + let () = fund_enforcer::(&mut enforcer_post_setup).await?; + Ok(enforcer_post_setup) +} + +async fn list_mempool_task( + bin_paths: BinPaths, + res_tx: mpsc::UnboundedSender>, +) -> anyhow::Result<()> { + let mut enforcer_post_setup = + setup(&bin_paths.others, res_tx.clone()).await?; + let mut sidechain = PostSetup::setup( + Init { + thunder_app: bin_paths.thunder()?.clone(), + data_dir_suffix: None, + }, + &enforcer_post_setup, + res_tx, + ) + .await?; + tracing::info!("Setup thunder node successfully"); + + tracing::debug!("Checking that a fresh mempool is empty"); + anyhow::ensure!(sidechain.rpc_client.list_mempool().await?.is_empty()); + + let deposit_address = sidechain.get_deposit_address().await?; + let () = deposit( + &mut enforcer_post_setup, + &mut sidechain, + &deposit_address, + DEPOSIT_AMOUNT, + DEPOSIT_FEE, + ) + .await?; + tracing::info!("Deposited to sidechain successfully"); + + tracing::debug!("Checking that a deposit alone leaves the mempool empty"); + anyhow::ensure!(sidechain.rpc_client.list_mempool().await?.is_empty()); + + let dest = sidechain.rpc_client.get_new_address().await?; + let txid = sidechain + .rpc_client + .create_transfer(dest, TRANSFER_AMOUNT, TRANSFER_FEE) + .await?; + tracing::info!(%txid, "Created a transfer"); + + tracing::debug!("Checking that the mempool holds the transfer"); + let mempool = sidechain.rpc_client.list_mempool().await?; + anyhow::ensure!(mempool.len() == 1); + let entry = &mempool[0]; + anyhow::ensure!(entry.txid == txid); + // The txid hashes over the canonical encoding, so both agree with the + // transaction the entry carries. + anyhow::ensure!(entry.tx.txid() == txid); + anyhow::ensure!(entry.size == entry.tx.canonical_size()); + anyhow::ensure!(entry.size > 0); + + tracing::debug!("Checking that a block empties the mempool"); + let () = sidechain.bmm_single(&mut enforcer_post_setup).await?; + anyhow::ensure!(sidechain.rpc_client.list_mempool().await?.is_empty()); + + drop(sidechain); + tracing::info!( + "Removing {}", + enforcer_post_setup.directories.base_dir.path().display() + ); + drop(enforcer_post_setup.tasks); + // Wait for tasks to die + sleep(std::time::Duration::from_secs(1)).await; + enforcer_post_setup.directories.base_dir.cleanup()?; + Ok(()) +} + +async fn list_mempool(bin_paths: BinPaths) -> anyhow::Result<()> { + let (res_tx, mut res_rx) = mpsc::unbounded(); + let _test_task: AbortOnDrop<()> = tokio::task::spawn({ + let res_tx = res_tx.clone(); + async move { + let res = list_mempool_task(bin_paths, res_tx.clone()).await; + let _send_err: Result<(), _> = res_tx.unbounded_send(res); + } + .in_current_span() + }) + .into(); + res_rx.next().await.ok_or_else(|| { + anyhow::anyhow!("Unexpected end of test task result stream") + })? +} + +pub fn list_mempool_trial( + bin_paths: BinPaths, + file_registry: TestFileRegistry, + failure_collector: TestFailureCollector, +) -> AsyncTrial>> { + AsyncTrial::new( + "list_mempool", + list_mempool(bin_paths).boxed(), + file_registry, + failure_collector, + ) +} diff --git a/integration_tests/main.rs b/integration_tests/main.rs index a8b80495..330aa1af 100644 --- a/integration_tests/main.rs +++ b/integration_tests/main.rs @@ -7,6 +7,7 @@ use tracing_subscriber::{filter as tracing_filter, layer::SubscriberExt}; mod block_template; mod ibd; mod integration_test; +mod list_mempool; mod setup; mod unknown_withdrawal; mod util; diff --git a/rpc-api/lib.rs b/rpc-api/lib.rs index 054a52be..eac23958 100644 --- a/rpc-api/lib.rs +++ b/rpc-api/lib.rs @@ -77,6 +77,16 @@ pub mod node { pub block_hash: Option, } + /// One transaction the mempool holds + #[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] + pub struct MempoolTx { + /// Blake3 over the canonical encoding + pub txid: Txid, + /// Canonical size in bytes + pub size: u64, + pub tx: Transaction, + } + #[open_api(ref_schemas[ Address, MerkleRoot, OutPoint, Output, OutputContent, Txid, schema::BitcoinTxid, thunder_schema::BitcoinAddr, @@ -164,6 +174,10 @@ pub mod node { &self, ) -> RpcResult>; + /// List the transactions the mempool holds, in no particular order. + #[method(name = "list_mempool")] + async fn list_mempool(&self) -> RpcResult>; + /// List peers #[method(name = "list_peers")] async fn list_peers(&self) -> RpcResult>;